Add CYB50 two-strategy backtrader comparison

backtrader-lab/cyb50_quant_two_strategies.py

@@ -0,0 +1,342 @@
+#!/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()

backtrader-lab/cyb50_quant_two_strategies_20260421-082038.json

@@ -0,0 +1,30 @@
+[
+  {
+    "strategy": "StandardBreakoutAtrStrategy",
+    "final_value": 169872.93,
+    "total_return_pct": 69.87,
+    "annual_return_pct": 4.76,
+    "sharpe": 0.261,
+    "max_drawdown_pct": 36.22,
+    "entries": 47,
+    "closed_trades": 47,
+    "win_rate_pct": 40.43,
+    "avg_exposure_pct": 18.34,
+    "first_bar": "2014-06-18",
+    "last_bar": "2026-04-13"
+  },
+  {
+    "strategy": "MinimalBreakoutTrailStrategy",
+    "final_value": 239640.48,
+    "total_return_pct": 139.64,
+    "annual_return_pct": 7.97,
+    "sharpe": 0.449,
+    "max_drawdown_pct": 30.81,
+    "entries": 39,
+    "closed_trades": 39,
+    "win_rate_pct": 38.46,
+    "avg_exposure_pct": 21.48,
+    "first_bar": "2014-06-18",
+    "last_bar": "2026-04-13"
+  }
+]

backtrader-lab/cyb50_quant_two_strategies_20260421-082038.md

@@ -0,0 +1,29 @@
+# CYB50 Two-Strategy Quant Backtest
+
+- Data: `chinext50.csv` (2014-06-18 to 2026-04-13, 2873 bars)
+- Initial cash: `100000`
+- Commission: `0.0010`
+- 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 |
+| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |
+| StandardBreakoutAtrStrategy | 169872.93 | 69.87% | 4.76% | 0.261 | 36.22% | 47 | 47 | 40.43% | 18.34% |
+| MinimalBreakoutTrailStrategy | 239640.48 | 139.64% | 7.97% | 0.449 | 30.81% | 39 | 39 | 38.46% | 21.48% |
+
+## Verdict
+
+- Higher return: **MinimalBreakoutTrailStrategy** (7.97% annual return, 30.81% max DD)
+- Lower drawdown: **MinimalBreakoutTrailStrategy** (30.81% max DD, 7.97% annual return)