Yuxuan Yan
69 lines
2.6 KiB
Python
69 lines
2.6 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
|
|
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_strategy(self, strategy_cls, **kwargs) -> None:
|
|
"""Add a strategy class to cerebro."""
|
|
self.cerebro.addstrategy(strategy_cls, **kwargs)
|
|
|
|
def setup(self, strategy_cls, strategy_kwargs: Optional[dict] = None) -> None:
|
|
"""Full setup: load data for all symbols, configure cerebro, add strategy."""
|
|
# Load data for all symbols
|
|
for sym in self.config.symbols:
|
|
self.add_data(sym)
|
|
|
|
# Configure cerebro
|
|
self.cerebro.broker.setcash(self.config.initial_cash)
|
|
self.cerebro.broker.setcommission(commission=self.config.commission)
|
|
|
|
# Add analyzers
|
|
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")
|
|
|
|
# Add strategy
|
|
strategy_kwargs = strategy_kwargs or {}
|
|
self.cerebro.addstrategy(strategy_cls, **strategy_kwargs)
|
|
|
|
def run(self, strategy_cls, strategy_kwargs: Optional[dict] = None) -> list:
|
|
"""Setup and run the backtest. Returns cerebro run results."""
|
|
self.setup(strategy_cls, strategy_kwargs)
|
|
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
|