#!/usr/bin/env python3 """Backtest two Chinext50 quant variants discussed in chat. Strategies 1. StandardBreakoutAtrStrategy - close > MA20 > MA60 - close breaks prior 20-day high - optional volume confirmation when usable volume exists - exits on close < MA20, close < prior 10-day low, trailing stop at highest_close_since_entry - 2.5 * ATR14, or profitable pullback below MA10 after 15% gain 2. MinimalBreakoutTrailStrategy - close > MA20 > MA60 - close breaks prior 20-day high - exits on close < MA20 or trailing drawdown > 8% Notes - The local chinext50 index CSV has volume=0 for all rows, so volume confirmation is automatically disabled instead of producing a permanently false rule. - Orders execute on next bar open, matching the usual Backtrader default. """ from __future__ import annotations import json from datetime import datetime from pathlib import Path import backtrader as bt import pandas as pd ROOT = Path(__file__).resolve().parent DATA_FILE = ROOT / "chinext50.csv" RUN_TS = datetime.now().strftime("%Y%m%d-%H%M%S") JSON_OUT = ROOT / f"cyb50_quant_two_strategies_{RUN_TS}.json" MD_OUT = ROOT / f"cyb50_quant_two_strategies_{RUN_TS}.md" INITIAL_CASH = 100000.0 COMMISSION = 0.001 TRADING_DAYS = 252 class Chinext50Data(bt.feeds.PandasData): params = ( ("datetime", None), ("open", "open"), ("high", "high"), ("low", "low"), ("close", "close"), ("volume", "volume"), ("openinterest", None), ) class BaseStrategy(bt.Strategy): params = (("printlog", False),) def __init__(self): self.order = None self.entry_count = 0 self.highest_close_since_entry = None self.exposure_sum = 0.0 self.exposure_bars = 0 def notify_order(self, order): if order.status in [order.Submitted, order.Accepted]: return if order.status == order.Completed and order.isbuy(): self.entry_count += 1 self.highest_close_since_entry = float(self.data.close[0]) self.order = None def next(self): portfolio = self.broker.getvalue() if portfolio > 0: exposure = abs(self.position.size) * float(self.data.close[0]) / portfolio self.exposure_sum += exposure self.exposure_bars += 1 if self.position: close = float(self.data.close[0]) if self.highest_close_since_entry is None: self.highest_close_since_entry = close else: self.highest_close_since_entry = max(self.highest_close_since_entry, close) else: self.highest_close_since_entry = None def avg_exposure(self) -> float: if self.exposure_bars == 0: return 0.0 return self.exposure_sum / self.exposure_bars def rebalance_to_weight(self, weight: float): weight = max(0.0, min(1.0, weight)) value = self.broker.getvalue() price = float(self.data.close[0]) if value <= 0 or price <= 0: return target_size = int(value * weight / price) delta = target_size - self.position.size if delta > 0: self.order = self.buy(size=delta) elif delta < 0: self.order = self.sell(size=abs(delta)) def go_long(self): self.rebalance_to_weight(1.0) def go_flat(self): if self.position: self.order = self.close() class StandardBreakoutAtrStrategy(BaseStrategy): params = ( ("ma_fast", 20), ("ma_slow", 60), ("breakout", 20), ("atr_period", 14), ("atr_mult", 2.5), ("exit_low_period", 10), ("profit_arm", 0.15), ("vol_period", 5), ("use_volume_filter", True), ) def __init__(self): super().__init__() close = self.data.close self.ma20 = bt.ind.SMA(close, period=self.p.ma_fast) self.ma60 = bt.ind.SMA(close, period=self.p.ma_slow) self.ma10 = bt.ind.SMA(close, period=10) self.atr = bt.ind.ATR(self.data, period=self.p.atr_period) self.highest_prev20 = bt.ind.Highest(self.data.high(-1), period=self.p.breakout) self.lowest_prev10 = bt.ind.Lowest(self.data.low(-1), period=self.p.exit_low_period) self.vol_ma5 = bt.ind.SMA(self.data.volume, period=self.p.vol_period) self.volume_filter_enabled = False def start(self): data_volume = getattr(self.data, "p", None) # enable volume filter only when the dataset has meaningful positive volume try: df = self.data._dataname if isinstance(df, pd.DataFrame) and "volume" in df.columns and (df["volume"] > 0).any(): self.volume_filter_enabled = bool(self.p.use_volume_filter) except Exception: self.volume_filter_enabled = False def next(self): super().next() if self.order: return if len(self) < max(self.p.ma_slow, self.p.breakout + 1, self.p.exit_low_period + 1, self.p.atr_period + 1): return close = float(self.data.close[0]) ma20 = float(self.ma20[0]) ma60 = float(self.ma60[0]) ma10 = float(self.ma10[0]) atr = float(self.atr[0]) highest_prev20 = float(self.highest_prev20[0]) lowest_prev10 = float(self.lowest_prev10[0]) volume_ok = True if self.volume_filter_enabled: volume_ok = float(self.data.volume[0]) > float(self.vol_ma5[0]) buy_signal = close > ma20 and ma20 > ma60 and close > highest_prev20 and volume_ok if not self.position and buy_signal: self.go_long() return if self.position: profit_pct = close / self.position.price - 1.0 if self.position.price else 0.0 trailing_stop = self.highest_close_since_entry - self.p.atr_mult * atr sell_signal = ( close < ma20 or close < lowest_prev10 or close < trailing_stop or (profit_pct > self.p.profit_arm and close < ma10) ) if sell_signal: self.go_flat() class MinimalBreakoutTrailStrategy(BaseStrategy): params = ( ("ma_fast", 20), ("ma_slow", 60), ("breakout", 20), ("trail_dd", 0.08), ) def __init__(self): super().__init__() close = self.data.close self.ma20 = bt.ind.SMA(close, period=self.p.ma_fast) self.ma60 = bt.ind.SMA(close, period=self.p.ma_slow) self.highest_prev20 = bt.ind.Highest(self.data.high(-1), period=self.p.breakout) def next(self): super().next() if self.order: return if len(self) < max(self.p.ma_slow, self.p.breakout + 1): return close = float(self.data.close[0]) ma20 = float(self.ma20[0]) ma60 = float(self.ma60[0]) highest_prev20 = float(self.highest_prev20[0]) buy_signal = close > ma20 and ma20 > ma60 and close > highest_prev20 if not self.position and buy_signal: self.go_long() return if self.position: trail_stop = self.highest_close_since_entry * (1.0 - self.p.trail_dd) if close < ma20 or close < trail_stop: self.go_flat() class ValueRecorder(bt.Analyzer): def start(self): self.values = [] def next(self): dt = self.strategy.datas[0].datetime.date(0) self.values.append((dt, self.strategy.broker.getvalue())) def get_analysis(self): return self.values def load_df() -> pd.DataFrame: return pd.read_csv(DATA_FILE, parse_dates=["datetime"]).set_index("datetime") def run_one(strategy_cls, df: pd.DataFrame) -> dict: cerebro = bt.Cerebro(stdstats=False) cerebro.broker.setcash(INITIAL_CASH) cerebro.broker.setcommission(commission=COMMISSION) cerebro.addstrategy(strategy_cls) cerebro.adddata(Chinext50Data(dataname=df)) cerebro.addanalyzer(bt.analyzers.Returns, _name="returns") cerebro.addanalyzer(bt.analyzers.DrawDown, _name="drawdown") cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name="sharpe", timeframe=bt.TimeFrame.Days, annualize=True, riskfreerate=0.02) cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name="trades") cerebro.addanalyzer(ValueRecorder, _name="values") results = cerebro.run() strat = results[0] returns = strat.analyzers.returns.get_analysis() dd = strat.analyzers.drawdown.get_analysis() sharpe = strat.analyzers.sharpe.get_analysis() trades = strat.analyzers.trades.get_analysis() value_series = strat.analyzers.values.get_analysis() closed = trades.get("total", {}).get("closed", 0) won = trades.get("won", {}).get("total", 0) win_rate = (won / closed * 100.0) if closed else 0.0 final_value = cerebro.broker.getvalue() return { "strategy": strategy_cls.__name__, "final_value": round(final_value, 2), "total_return_pct": round((final_value / INITIAL_CASH - 1.0) * 100.0, 2), "annual_return_pct": round(returns.get("rnorm100", 0.0), 2), "sharpe": None if sharpe.get("sharperatio") is None else round(sharpe.get("sharperatio"), 3), "max_drawdown_pct": round(dd.get("max", {}).get("drawdown", 0.0), 2), "entries": strat.entry_count, "closed_trades": closed, "win_rate_pct": round(win_rate, 2), "avg_exposure_pct": round(strat.avg_exposure() * 100.0, 2), "first_bar": value_series[0][0].isoformat() if value_series else None, "last_bar": value_series[-1][0].isoformat() if value_series else None, } def build_markdown(results: list[dict], df: pd.DataFrame) -> str: lines = [ "# CYB50 Two-Strategy Quant Backtest", "", f"- Data: `{DATA_FILE.name}` ({df.index.min().date()} to {df.index.max().date()}, {len(df)} bars)", f"- Initial cash: `{INITIAL_CASH:.0f}`", f"- Commission: `{COMMISSION:.4f}`", "- Note: local index CSV has `volume=0` on all rows, so volume confirmation was auto-disabled for the standard strategy.", "", "## Rules", "", "### StandardBreakoutAtrStrategy", "- Entry: close > MA20 > MA60 and close > prior 20-day high", "- Extra filter: volume > 5-day average only when usable volume exists", "- Exit: close < MA20, close < prior 10-day low, close < highest close since entry - 2.5 * ATR14, or after 15% gain a close < MA10", "", "### MinimalBreakoutTrailStrategy", "- Entry: close > MA20 > MA60 and close > prior 20-day high", "- Exit: close < MA20 or trailing drawdown from highest close since entry > 8%", "", "## Results", "", "| Strategy | Final Value | Total Return | Annual Return | Sharpe | Max DD | Entries | Closed Trades | Win Rate | Avg Exposure |", "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |", ] for row in results: lines.append( f"| {row['strategy']} | {row['final_value']:.2f} | {row['total_return_pct']:.2f}% | {row['annual_return_pct']:.2f}% | " f"{('n/a' if row['sharpe'] is None else row['sharpe'])} | {row['max_drawdown_pct']:.2f}% | {row['entries']} | " f"{row['closed_trades']} | {row['win_rate_pct']:.2f}% | {row['avg_exposure_pct']:.2f}% |" ) best_return = max(results, key=lambda x: x['annual_return_pct']) best_defensive = min(results, key=lambda x: x['max_drawdown_pct']) lines += [ "", "## Verdict", "", f"- Higher return: **{best_return['strategy']}** ({best_return['annual_return_pct']:.2f}% annual return, {best_return['max_drawdown_pct']:.2f}% max DD)", f"- Lower drawdown: **{best_defensive['strategy']}** ({best_defensive['max_drawdown_pct']:.2f}% max DD, {best_defensive['annual_return_pct']:.2f}% annual return)", ] return "\n".join(lines) + "\n" def main(): df = load_df() results = [ run_one(StandardBreakoutAtrStrategy, df.copy()), run_one(MinimalBreakoutTrailStrategy, df.copy()), ] JSON_OUT.write_text(json.dumps(results, ensure_ascii=False, indent=2), encoding="utf-8") MD_OUT.write_text(build_markdown(results, df), encoding="utf-8") print(f"JSON: {JSON_OUT}") print(f"MD: {MD_OUT}") print(json.dumps(results, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()