Skip to content

Refactored optimization & backtest modules #25

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 5 commits into
base: main
Choose a base branch
from
Open

Refactored optimization & backtest modules #25

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
3b211d0
Refactored optimization & backtest modules
rrawatt Mar 5, 2025
68cb685
Update portfolio.py
rrawatt Mar 7, 2025
f48cf37
Update selection.py
rrawatt Mar 11, 2025
4318d92
Update helper_functions.py
rrawatt Apr 4, 2025
307e445
Update constraints.py
rrawatt Apr 4, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
99 changes: 39 additions & 60 deletions src/backtest.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,21 +1,17 @@
'''
PorQua : a python library for portfolio optimization and backtesting
PorQua is part of GeomScale project
PorQua: A Python Library for Portfolio Optimization and Backtesting
Part of the GeomScale project

Copyright (c) 2024 Cyril Bachelard
Copyright (c) 2024 Minh Ha Ho

Licensed under GNU LGPL.3, see LICENCE file
Licensed under GNU LGPL.3; see LICENCE file.
'''


############################################################################
### CLASSES BacktestData, BacktestService, Backtest
############################################################################




import os
from typing import Optional
import pickle
Expand All @@ -30,17 +26,12 @@
from builders import SelectionItemBuilder, OptimizationItemBuilder





class BacktestData():

def __init__(self):
pass


class BacktestService():

def __init__(self,
data: BacktestData,
selection_item_builders: dict[str, SelectionItemBuilder],
Expand All @@ -54,7 +45,7 @@ def __init__(self,
self.optimization_item_builders = optimization_item_builders
self.settings = settings if settings is not None else {}
self.settings.update(kwargs)
# Initialize the selection and optimization data

self.selection = Selection()
self.optimization_data = OptimizationData([])

Expand Down Expand Up @@ -135,23 +126,19 @@ def build_selection(self, rebdate: str) -> None:

def build_optimization(self, rebdate: str) -> None:

# Initialize the optimization constraints
self.optimization.constraints = Constraints(selection = self.selection.selected)
self.optimization.constraints = Constraints(selection=self.selection.selected)

# Loop over the optimization_item_builders items
for item_builder in self.optimization_item_builders.values():
item_builder(self, rebdate)
return None

def prepare_rebalancing(self, rebalancing_date: str) -> None:
self.build_selection(rebdate = rebalancing_date)
self.build_optimization(rebdate = rebalancing_date)
self.build_selection(rebdate=rebalancing_date)
self.build_optimization(rebdate=rebalancing_date)
return None



class Backtest:

def __init__(self) -> None:
self._strategy = Strategy([])
self._output = {}
Expand All @@ -165,16 +152,16 @@ def output(self):
return self._output

def append_output(self,
date_key = None,
output_key = None,
value = None):
date_key=None,
output_key=None,
value=None):
if value is None:
return True

if date_key in self.output.keys():
if output_key in self.output[date_key].keys():
raise Warning(f"Output key '{output_key}' for date key '{date_key}' \
already exists and will be overwritten.")
raise Warning(f"Output key '{output_key}' for date key '{date_key}' "
"already exists and will be overwritten.")
self.output[date_key][output_key] = value
else:
self.output[date_key] = {}
Expand All @@ -185,41 +172,37 @@ def append_output(self,
def rebalance(self,
bs: BacktestService,
rebalancing_date: str) -> None:

# Prepare the rebalancing, i.e., the optimization problem
bs.prepare_rebalancing(rebalancing_date = rebalancing_date)
bs.prepare_rebalancing(rebalancing_date=rebalancing_date)

# Solve the optimization problem
try:
bs.optimization.set_objective(optimization_data = bs.optimization_data)
bs.optimization.set_objective(optimization_data=bs.optimization_data)
bs.optimization.solve()
except Exception as error:
raise RuntimeError(error)

return None

def run(self, bs: BacktestService) -> None:

for rebalancing_date in bs.settings['rebdates']:

if not bs.settings.get('quiet'):
print(f'Rebalancing date: {rebalancing_date}')

self.rebalance(bs = bs,
rebalancing_date = rebalancing_date)
self.rebalance(bs=bs,
rebalancing_date=rebalancing_date)

# Append portfolio to strategy
weights = bs.optimization.results['weights']
portfolio = Portfolio(rebalancing_date = rebalancing_date, weights = weights)
portfolio = Portfolio(rebalancing_date=rebalancing_date, weights=weights)
self.strategy.portfolios.append(portfolio)

# Append stuff to output if a custom append function is provided
# Append additional output if a custom append function is provided
append_fun = bs.settings.get('append_fun')
if append_fun is not None:
append_fun(backtest = self,
bs = bs,
rebalancing_date = rebalancing_date,
what = bs.settings.get('append_fun_args'))
append_fun(backtest=self,
bs=bs,
rebalancing_date=rebalancing_date,
what=bs.settings.get('append_fun_args'))

return None

Expand All @@ -228,7 +211,7 @@ def save(self,
path: Optional[str] = None) -> None:
try:
if path is not None and filename is not None:
filename = os.path.join(path, filename) #// alternatively, use pathlib package
filename = os.path.join(path, filename) # Alternatively, use pathlib package
with open(filename, "wb") as f:
pickle.dump(self, f, protocol=pickle.HIGHEST_PROTOCOL)
except Exception as ex:
Expand All @@ -237,34 +220,30 @@ def save(self,
return None



# --------------------------------------------------------------------------
# Helper functions
# --------------------------------------------------------------------------

def append_custom(backtest: Backtest,
bs: BacktestService,
rebalancing_date: Optional[str] = None,
what: Optional[list] = None) -> None:

if what is None:
what = ['w_dict', 'objective']
what = ['portfolio_ids', 'objective']

for key in what:
if key == 'w_dict':
w_dict = bs.optimization.results['w_dict']
for key in w_dict.keys():
weights = w_dict[key]
if hasattr(weights, 'to_dict'):
weights = weights.to_dict()
portfolio = Portfolio(rebalancing_date = rebalancing_date, weights = weights)
backtest.append_output(date_key = rebalancing_date,
output_key = f'weights_{key}',
value = pd.Series(portfolio.weights))
if key == 'portfolio_ids':
portfolio_ids = bs.optimization.results.get('portfolio_ids', {})
for pid, indices in portfolio_ids.items():

weights = bs.optimization.results.get('weights', {})

segment_weights = {i: weights[i] for i in indices if i in weights}
portfolio = Portfolio(rebalancing_date=rebalancing_date, weights=segment_weights)
backtest.append_output(date_key=rebalancing_date,
output_key=f'weights_portfolio_{pid}',
value=pd.Series(portfolio.weights))
else:
if not key in bs.optimization.results.keys():
if key not in bs.optimization.results.keys():
continue
backtest.append_output(date_key = rebalancing_date,
output_key = key,
value = bs.optimization.results[key])
backtest.append_output(date_key=rebalancing_date,
output_key=key,
value=bs.optimization.results[key])
return None
Loading