cyb50-quant/backtrader-lab/cyb50_quant_two_strategies.py

#!/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()