Yuxuan Yan
85 lines
3.5 KiB
Python
85 lines
3.5 KiB
Python
"""BacktestRunner: orchestrates data loading, cerebro setup, and execution."""
|
|
import logging
|
|
import backtrader as bt
|
|
from typing import Optional
|
|
|
|
from backtest.config import BacktestConfig
|
|
from backtest.feed import df_to_bt_feed, df_to_signal_feed
|
|
from data.downloader import download_daily
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class BacktestRunner:
|
|
"""Run backtrader backtests with Chinese equity data."""
|
|
|
|
def __init__(self, config: BacktestConfig):
|
|
self.config = config
|
|
self.cerebro = bt.Cerebro()
|
|
self._results: Optional[list] = None
|
|
|
|
def add_data(self, symbol: str) -> None:
|
|
"""Download data for a symbol and add to cerebro as a feed."""
|
|
df = download_daily(
|
|
symbol=symbol,
|
|
start=self.config.start_date,
|
|
end=self.config.end_date,
|
|
adjust=self.config.adjust,
|
|
)
|
|
feed = df_to_bt_feed(df)
|
|
self.cerebro.adddata(feed, name=symbol)
|
|
logger.info(f"Added {symbol}: {len(df)} bars")
|
|
|
|
def add_signal_data(self, df, name: str) -> None:
|
|
"""Add a pre-built OHLCV+signal DataFrame as a SignalPandasData feed."""
|
|
feed = df_to_signal_feed(df)
|
|
self.cerebro.adddata(feed, name=name)
|
|
logger.info(f"Added signal feed {name}: {len(df)} bars")
|
|
|
|
def add_strategy(self, strategy_cls, **kwargs) -> None:
|
|
"""Add a strategy class to cerebro."""
|
|
self.cerebro.addstrategy(strategy_cls, **kwargs)
|
|
|
|
def _configure(self) -> None:
|
|
"""Configure broker, sizer, and analyzers (independent of data feeds)."""
|
|
self.cerebro.broker.setcash(self.config.initial_cash)
|
|
self.cerebro.broker.setcommission(commission=self.config.commission)
|
|
self.cerebro.addsizer(bt.sizers.PercentSizer, percents=self.config.sizer_percent * 100)
|
|
|
|
self.cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name="sharpe", riskfreerate=0.02)
|
|
self.cerebro.addanalyzer(bt.analyzers.DrawDown, _name="drawdown")
|
|
self.cerebro.addanalyzer(bt.analyzers.Returns, _name="returns")
|
|
self.cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name="trades")
|
|
self.cerebro.addanalyzer(
|
|
bt.analyzers.TimeReturn, _name="timereturn", timeframe=bt.TimeFrame.Days
|
|
)
|
|
|
|
def setup(self, strategy_cls, strategy_kwargs: Optional[dict] = None) -> None:
|
|
"""Full setup: load data for all symbols, configure cerebro, add strategy."""
|
|
for sym in self.config.symbols:
|
|
self.add_data(sym)
|
|
self._configure()
|
|
self.cerebro.addstrategy(strategy_cls, **(strategy_kwargs or {}))
|
|
|
|
def run(self, strategy_cls, strategy_kwargs: Optional[dict] = None) -> list:
|
|
"""Setup (downloading all symbols) and run the backtest."""
|
|
self.setup(strategy_cls, strategy_kwargs)
|
|
return self._execute()
|
|
|
|
def run_prepared(self, strategy_cls, strategy_kwargs: Optional[dict] = None) -> list:
|
|
"""Run a backtest using feeds already added via ``add_signal_data``."""
|
|
self._configure()
|
|
self.cerebro.addstrategy(strategy_cls, **(strategy_kwargs or {}))
|
|
return self._execute()
|
|
|
|
def _execute(self) -> list:
|
|
start_val = self.cerebro.broker.getvalue()
|
|
logger.info(f"Starting portfolio value: {start_val:,.2f}")
|
|
self._results = self.cerebro.run()
|
|
end_val = self.cerebro.broker.getvalue()
|
|
logger.info(f"Ending portfolio value: {end_val:,.2f}")
|
|
return self._results
|
|
|
|
def get_results(self) -> Optional[list]:
|
|
return self._results
|