cyb50-quant/backtrader-lab/chinext50_experiments.py

#!/usr/bin/env python3
"""Run Chinext50 Backtrader experiments and save a short report."""

from __future__ import annotations

import argparse
import itertools
import math
from pathlib import Path

import backtrader as bt
import pandas as pd


ROOT = Path(__file__).resolve().parent
DATA_FILE = ROOT / "chinext50.csv"
REPORT_FILE = ROOT / "chinext50_experiment_summary.md"
DUALTHRUST_OPT_REPORT_FILE = ROOT / "chinext50_dualthrust_optimization.md"
INITIAL_CASH = 100000.0
COMMISSION = 0.001
TRADING_DAYS = 252


class Chinext50Data(bt.feeds.PandasData):
    """Explicit column mapping for the local Chinext50 CSV."""

    params = (
        ("datetime", None),
        ("open", "open"),
        ("high", "high"),
        ("low", "low"),
        ("close", "close"),
        ("volume", "volume"),
        ("openinterest", None),
    )


class BaseIndexStrategy(bt.Strategy):
    """Common helpers for long-only index timing strategies."""

    def __init__(self):
        self.order = None
        self.entry_count = 0
        self.bars_in_market = 0
        self.exposure_sum = 0.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.order = None

    def next(self):
        portfolio_value = self.broker.getvalue()
        if portfolio_value > 0:
            position_value = abs(self.position.size) * self.data.close[0]
            exposure = position_value / portfolio_value
            self.exposure_sum += exposure
            if exposure > 0:
                self.bars_in_market += 1

    def _target_size_for_weight(self, target_weight: float) -> int:
        target_weight = max(0.0, target_weight)
        portfolio_value = self.broker.getvalue()
        price = self.data.close[0]
        if portfolio_value <= 0 or price <= 0:
            return 0
        target_value = portfolio_value * target_weight
        return max(int(target_value / price), 0)

    def _rebalance_to_weight(self, target_weight: float):
        target_size = self._target_size_for_weight(target_weight)
        current_size = self.position.size
        size_delta = target_size - current_size
        if size_delta > 0:
            self.order = self.buy(size=size_delta)
        elif size_delta < 0:
            self.order = self.sell(size=abs(size_delta))

    def _buy_full(self):
        self._rebalance_to_weight(1.0)

    def _go_flat(self):
        if self.position:
            self.order = self.close()


class SuperTrendIndicator(bt.Indicator):
    """Minimal SuperTrend implementation built from Backtrader ATR."""

    lines = ("supertrend",)
    params = (
        ("period", 10),
        ("multiplier", 3.0),
    )
    plotinfo = {"subplot": False}

    def __init__(self):
        hl2 = (self.data.high + self.data.low) / 2.0
        self.atr = bt.indicators.ATR(self.data, period=self.p.period)
        self.basic_upper = hl2 + self.p.multiplier * self.atr
        self.basic_lower = hl2 - self.p.multiplier * self.atr
        self._final_upper = None
        self._final_lower = None

    def next(self):
        if math.isnan(self.atr[0]):
            return

        basic_upper = float(self.basic_upper[0])
        basic_lower = float(self.basic_lower[0])
        close = float(self.data.close[0])

        if self._final_upper is None or self._final_lower is None:
            self._final_upper = basic_upper
            self._final_lower = basic_lower
            self.lines.supertrend[0] = basic_lower
            return

        prev_final_upper = self._final_upper
        prev_final_lower = self._final_lower
        prev_close = float(self.data.close[-1])
        prev_supertrend = float(self.lines.supertrend[-1])

        final_upper = basic_upper if basic_upper < prev_final_upper or prev_close > prev_final_upper else prev_final_upper
        final_lower = basic_lower if basic_lower > prev_final_lower or prev_close < prev_final_lower else prev_final_lower

        if prev_supertrend == prev_final_upper:
            supertrend = final_upper if close <= final_upper else final_lower
        else:
            supertrend = final_lower if close >= final_lower else final_upper

        self._final_upper = final_upper
        self._final_lower = final_lower
        self.lines.supertrend[0] = supertrend


class TrendRegimeFlatStrategy(BaseIndexStrategy):
    """
    Trend following with flat/cash regime control.

    Rules:
    - Long only when short trend > medium trend and price stays above a regime MA
    - Exit fully when the trend breaks or short-term volatility spikes above a cap
    """

    params = (
        ("fast", 20),
        ("slow", 60),
        ("regime", 120),
        ("vol_fast", 20),
        ("vol_slow", 60),
        ("vol_cap", 1.10),
    )

    def __init__(self):
        super().__init__()
        close = self.data.close
        self.sma_fast = bt.indicators.SMA(close, period=self.p.fast)
        self.sma_slow = bt.indicators.SMA(close, period=self.p.slow)
        self.sma_regime = bt.indicators.SMA(close, period=self.p.regime)
        returns = bt.indicators.PctChange(close, period=1)
        self.vol_fast = bt.indicators.StdDev(returns, period=self.p.vol_fast)
        self.vol_slow = bt.indicators.StdDev(returns, period=self.p.vol_slow)

    def next(self):
        super().next()
        if self.order:
            return
        if (
            math.isnan(self.sma_fast[0])
            or math.isnan(self.sma_slow[0])
            or math.isnan(self.sma_regime[0])
            or math.isnan(self.vol_fast[0])
            or math.isnan(self.vol_slow[0])
            or self.vol_slow[0] == 0
        ):
            return

        bullish_trend = self.sma_fast[0] > self.sma_slow[0] and self.data.close[0] > self.sma_regime[0]
        calm_enough = self.vol_fast[0] <= self.vol_slow[0] * self.p.vol_cap

        if bullish_trend and calm_enough and not self.position:
            self._buy_full()
        elif self.position and (not bullish_trend or not calm_enough):
            self._go_flat()


class TrendTightVolStrategy(BaseIndexStrategy):
    """Trend following with the same regime logic but a tighter volatility cap."""

    params = (
        ("fast", 20),
        ("slow", 60),
        ("regime", 120),
        ("vol_fast", 20),
        ("vol_slow", 60),
        ("vol_cap", 0.95),
    )

    def __init__(self):
        super().__init__()
        close = self.data.close
        self.sma_fast = bt.indicators.SMA(close, period=self.p.fast)
        self.sma_slow = bt.indicators.SMA(close, period=self.p.slow)
        self.sma_regime = bt.indicators.SMA(close, period=self.p.regime)
        returns = bt.indicators.PctChange(close, period=1)
        self.vol_fast = bt.indicators.StdDev(returns, period=self.p.vol_fast)
        self.vol_slow = bt.indicators.StdDev(returns, period=self.p.vol_slow)

    def next(self):
        super().next()
        if self.order:
            return
        if (
            math.isnan(self.sma_fast[0])
            or math.isnan(self.sma_slow[0])
            or math.isnan(self.sma_regime[0])
            or math.isnan(self.vol_fast[0])
            or math.isnan(self.vol_slow[0])
            or self.vol_slow[0] <= 0
        ):
            return

        bullish_trend = self.sma_fast[0] > self.sma_slow[0] and self.data.close[0] > self.sma_regime[0]
        calm_enough = self.vol_fast[0] <= self.vol_slow[0] * self.p.vol_cap

        if bullish_trend and calm_enough and not self.position:
            self._buy_full()
        elif self.position and (not bullish_trend or not calm_enough):
            self._go_flat()


class TrendLooseVolStrategy(BaseIndexStrategy):
    """Trend following with the same regime logic but a looser volatility cap."""

    params = (
        ("fast", 20),
        ("slow", 60),
        ("regime", 120),
        ("vol_fast", 20),
        ("vol_slow", 60),
        ("vol_cap", 1.25),
    )

    def __init__(self):
        super().__init__()
        close = self.data.close
        self.sma_fast = bt.indicators.SMA(close, period=self.p.fast)
        self.sma_slow = bt.indicators.SMA(close, period=self.p.slow)
        self.sma_regime = bt.indicators.SMA(close, period=self.p.regime)
        returns = bt.indicators.PctChange(close, period=1)
        self.vol_fast = bt.indicators.StdDev(returns, period=self.p.vol_fast)
        self.vol_slow = bt.indicators.StdDev(returns, period=self.p.vol_slow)

    def next(self):
        super().next()
        if self.order:
            return
        if (
            math.isnan(self.sma_fast[0])
            or math.isnan(self.sma_slow[0])
            or math.isnan(self.sma_regime[0])
            or math.isnan(self.vol_fast[0])
            or math.isnan(self.vol_slow[0])
            or self.vol_slow[0] <= 0
        ):
            return

        bullish_trend = self.sma_fast[0] > self.sma_slow[0] and self.data.close[0] > self.sma_regime[0]
        calm_enough = self.vol_fast[0] <= self.vol_slow[0] * self.p.vol_cap

        if bullish_trend and calm_enough and not self.position:
            self._buy_full()
        elif self.position and (not bullish_trend or not calm_enough):
            self._go_flat()


class SmaLongFilterTrendStrategy(BaseIndexStrategy):
    """Simple SMA trend following with a long-MA regime filter."""

    params = (
        ("fast", 20),
        ("slow", 60),
        ("regime", 120),
    )

    def __init__(self):
        super().__init__()
        close = self.data.close
        self.sma_fast = bt.indicators.SMA(close, period=self.p.fast)
        self.sma_slow = bt.indicators.SMA(close, period=self.p.slow)
        self.sma_regime = bt.indicators.SMA(close, period=self.p.regime)

    def next(self):
        super().next()
        if self.order:
            return
        if math.isnan(self.sma_fast[0]) or math.isnan(self.sma_slow[0]) or math.isnan(self.sma_regime[0]):
            return

        signal = self.sma_fast[0] > self.sma_slow[0] and self.data.close[0] > self.sma_regime[0]

        if signal and not self.position:
            self._buy_full()
        elif self.position and not signal:
            self._go_flat()


class MomentumRegimeStrategy(BaseIndexStrategy):
    """
    Single-asset momentum timing proxy for an index.

    Rules:
    - Long only when both short- and medium-term momentum are positive
    - Require price above a longer regime MA to avoid deep bear phases
    - Otherwise stay flat
    """

    params = (
        ("mom_short", 20),
        ("mom_long", 120),
        ("regime", 150),
    )

    def __init__(self):
        super().__init__()
        close = self.data.close
        self.roc_short = bt.indicators.ROC(close, period=self.p.mom_short)
        self.roc_long = bt.indicators.ROC(close, period=self.p.mom_long)
        self.sma_regime = bt.indicators.SMA(close, period=self.p.regime)

    def next(self):
        super().next()
        if self.order:
            return
        if math.isnan(self.roc_short[0]) or math.isnan(self.roc_long[0]) or math.isnan(self.sma_regime[0]):
            return

        signal = (
            self.roc_short[0] > 0
            and self.roc_long[0] > 0
            and self.data.close[0] > self.sma_regime[0]
        )

        if signal and not self.position:
            self._buy_full()
        elif self.position and not signal:
            self._go_flat()


class MomentumBasicStrategy(BaseIndexStrategy):
    """Dual-window momentum without the long regime filter."""

    params = (
        ("mom_short", 20),
        ("mom_long", 120),
    )

    def __init__(self):
        super().__init__()
        close = self.data.close
        self.roc_short = bt.indicators.ROC(close, period=self.p.mom_short)
        self.roc_long = bt.indicators.ROC(close, period=self.p.mom_long)

    def next(self):
        super().next()
        if self.order:
            return
        if math.isnan(self.roc_short[0]) or math.isnan(self.roc_long[0]):
            return

        signal = self.roc_short[0] > 0 and self.roc_long[0] > 0

        if signal and not self.position:
            self._buy_full()
        elif self.position and not signal:
            self._go_flat()


class MomentumMaFilterStrategy(BaseIndexStrategy):
    """Momentum timing with a medium-term moving-average confirmation filter."""

    params = (
        ("mom_short", 20),
        ("mom_long", 120),
        ("ma_filter", 60),
    )

    def __init__(self):
        super().__init__()
        close = self.data.close
        self.roc_short = bt.indicators.ROC(close, period=self.p.mom_short)
        self.roc_long = bt.indicators.ROC(close, period=self.p.mom_long)
        self.sma_filter = bt.indicators.SMA(close, period=self.p.ma_filter)

    def next(self):
        super().next()
        if self.order:
            return
        if math.isnan(self.roc_short[0]) or math.isnan(self.roc_long[0]) or math.isnan(self.sma_filter[0]):
            return

        signal = (
            self.roc_short[0] > 0
            and self.roc_long[0] > 0
            and self.data.close[0] > self.sma_filter[0]
        )

        if signal and not self.position:
            self._buy_full()
        elif self.position and not signal:
            self._go_flat()


class MomentumDefensiveFilterStrategy(BaseIndexStrategy):
    """Momentum timing with a regime filter and simple fast/slow volatility cap."""

    params = (
        ("mom_short", 20),
        ("mom_long", 120),
        ("regime", 150),
        ("vol_fast", 20),
        ("vol_slow", 60),
        ("vol_cap", 1.05),
    )

    def __init__(self):
        super().__init__()
        close = self.data.close
        self.roc_short = bt.indicators.ROC(close, period=self.p.mom_short)
        self.roc_long = bt.indicators.ROC(close, period=self.p.mom_long)
        self.sma_regime = bt.indicators.SMA(close, period=self.p.regime)
        returns = bt.indicators.PctChange(close, period=1)
        self.vol_fast = bt.indicators.StdDev(returns, period=self.p.vol_fast)
        self.vol_slow = bt.indicators.StdDev(returns, period=self.p.vol_slow)

    def next(self):
        super().next()
        if self.order:
            return
        if (
            math.isnan(self.roc_short[0])
            or math.isnan(self.roc_long[0])
            or math.isnan(self.sma_regime[0])
            or math.isnan(self.vol_fast[0])
            or math.isnan(self.vol_slow[0])
            or self.vol_slow[0] <= 0
        ):
            return

        signal = (
            self.roc_short[0] > 0
            and self.roc_long[0] > 0
            and self.data.close[0] > self.sma_regime[0]
            and self.vol_fast[0] <= self.vol_slow[0] * self.p.vol_cap
        )

        if signal and not self.position:
            self._buy_full()
        elif self.position and not signal:
            self._go_flat()


class MomentumAtrTrailBasicStrategy(BaseIndexStrategy):
    """Momentum entry with ATR trailing exit but no regime filter."""

    params = (
        ("mom_short", 20),
        ("mom_long", 120),
        ("atr_period", 20),
        ("atr_mult", 4.0),
    )

    def __init__(self):
        super().__init__()
        close = self.data.close
        self.roc_short = bt.indicators.ROC(close, period=self.p.mom_short)
        self.roc_long = bt.indicators.ROC(close, period=self.p.mom_long)
        self.atr = bt.indicators.ATR(self.data, period=self.p.atr_period)
        self.highest_close = None

    def notify_order(self, order):
        super().notify_order(order)
        if order.status != order.Completed:
            return
        if order.isbuy():
            self.highest_close = order.executed.price
        elif not self.position:
            self.highest_close = None

    def next(self):
        super().next()
        if self.order:
            return
        if math.isnan(self.roc_short[0]) or math.isnan(self.roc_long[0]) or math.isnan(self.atr[0]):
            return

        entry_signal = self.roc_short[0] > 0 and self.roc_long[0] > 0
        if not self.position:
            if entry_signal:
                self._buy_full()
            return

        if self.highest_close is None:
            self.highest_close = self.data.close[0]
        self.highest_close = max(self.highest_close, self.data.close[0])
        trailing_stop = self.highest_close - self.p.atr_mult * self.atr[0]

        if self.data.close[0] < trailing_stop or not entry_signal:
            self._go_flat()


class MomentumAtrTrailStrategy(BaseIndexStrategy):
    """Momentum entry with a regime filter and ATR trailing exit."""

    params = (
        ("mom_short", 20),
        ("mom_long", 120),
        ("regime", 150),
        ("atr_period", 20),
        ("atr_mult", 4.0),
    )

    def __init__(self):
        super().__init__()
        close = self.data.close
        self.roc_short = bt.indicators.ROC(close, period=self.p.mom_short)
        self.roc_long = bt.indicators.ROC(close, period=self.p.mom_long)
        self.sma_regime = bt.indicators.SMA(close, period=self.p.regime)
        self.atr = bt.indicators.ATR(self.data, period=self.p.atr_period)
        self.highest_close = None

    def notify_order(self, order):
        super().notify_order(order)
        if order.status != order.Completed:
            return
        if order.isbuy():
            self.highest_close = order.executed.price
        elif not self.position:
            self.highest_close = None

    def next(self):
        super().next()
        if self.order:
            return
        if (
            math.isnan(self.roc_short[0])
            or math.isnan(self.roc_long[0])
            or math.isnan(self.sma_regime[0])
            or math.isnan(self.atr[0])
        ):
            return

        entry_signal = (
            self.roc_short[0] > 0
            and self.roc_long[0] > 0
            and self.data.close[0] > self.sma_regime[0]
        )
        if not self.position:
            if entry_signal:
                self._buy_full()
            return

        if self.highest_close is None:
            self.highest_close = self.data.close[0]
        self.highest_close = max(self.highest_close, self.data.close[0])
        trailing_stop = self.highest_close - self.p.atr_mult * self.atr[0]

        if self.data.close[0] < trailing_stop or not entry_signal:
            self._go_flat()


class DonchianRegimeStrategy(BaseIndexStrategy):
    """Donchian breakout gated by a long moving-average regime filter."""

    params = (
        ("breakout", 55),
        ("exit_period", 30),
        ("regime", 150),
    )

    def __init__(self):
        super().__init__()
        self.highest_high = bt.indicators.Highest(self.data.high, period=self.p.breakout)
        self.lowest_low = bt.indicators.Lowest(self.data.low, period=self.p.exit_period)
        self.sma_regime = bt.indicators.SMA(self.data.close, period=self.p.regime)

    def next(self):
        super().next()
        if self.order:
            return
        if len(self) <= max(self.p.breakout, self.p.exit_period, self.p.regime):
            return
        if math.isnan(self.sma_regime[0]) or math.isnan(self.highest_high[-1]) or math.isnan(self.lowest_low[-1]):
            return

        breakout_signal = self.data.close[0] > self.highest_high[-1] and self.data.close[0] > self.sma_regime[0]
        exit_signal = self.data.close[0] < self.lowest_low[-1] or self.data.close[0] < self.sma_regime[0]

        if breakout_signal and not self.position:
            self._buy_full()
        elif self.position and exit_signal:
            self._go_flat()


class DonchianBasicStrategy(BaseIndexStrategy):
    """Pure Donchian breakout without regime or ADX filtering."""

    params = (
        ("breakout", 55),
        ("exit_period", 30),
    )

    def __init__(self):
        super().__init__()
        self.highest_high = bt.indicators.Highest(self.data.high, period=self.p.breakout)
        self.lowest_low = bt.indicators.Lowest(self.data.low, period=self.p.exit_period)

    def next(self):
        super().next()
        if self.order:
            return
        if len(self) <= max(self.p.breakout, self.p.exit_period):
            return
        if math.isnan(self.highest_high[-1]) or math.isnan(self.lowest_low[-1]):
            return

        breakout_signal = self.data.close[0] > self.highest_high[-1]
        exit_signal = self.data.close[0] < self.lowest_low[-1]

        if breakout_signal and not self.position:
            self._buy_full()
        elif self.position and exit_signal:
            self._go_flat()


class DonchianAdxStrategy(BaseIndexStrategy):
    """Donchian breakout with an ADX trend-strength filter."""

    params = (
        ("breakout", 55),
        ("exit_period", 30),
        ("adx_period", 14),
        ("adx_threshold", 20.0),
    )

    def __init__(self):
        super().__init__()
        self.highest_high = bt.indicators.Highest(self.data.high, period=self.p.breakout)
        self.lowest_low = bt.indicators.Lowest(self.data.low, period=self.p.exit_period)
        self.adx = bt.indicators.ADX(self.data, period=self.p.adx_period)

    def next(self):
        super().next()
        if self.order:
            return
        if len(self) <= max(self.p.breakout, self.p.exit_period, self.p.adx_period):
            return
        if math.isnan(self.highest_high[-1]) or math.isnan(self.lowest_low[-1]) or math.isnan(self.adx[0]):
            return

        breakout_signal = self.data.close[0] > self.highest_high[-1] and self.adx[0] >= self.p.adx_threshold
        exit_signal = self.data.close[0] < self.lowest_low[-1]

        if breakout_signal and not self.position:
            self._buy_full()
        elif self.position and exit_signal:
            self._go_flat()


class DonchianAtrTrailStrategy(BaseIndexStrategy):
    """Donchian breakout with ATR trailing protection but no vol-target overlay."""

    params = (
        ("breakout", 55),
        ("exit_period", 30),
        ("atr_period", 20),
        ("atr_mult", 4.0),
    )

    def __init__(self):
        super().__init__()
        self.highest_high = bt.indicators.Highest(self.data.high, period=self.p.breakout)
        self.lowest_low = bt.indicators.Lowest(self.data.low, period=self.p.exit_period)
        self.atr = bt.indicators.ATR(self.data, period=self.p.atr_period)
        self.highest_close = None

    def notify_order(self, order):
        super().notify_order(order)
        if order.status != order.Completed:
            return
        if order.isbuy():
            self.highest_close = order.executed.price
        elif not self.position:
            self.highest_close = None

    def next(self):
        super().next()
        if self.order:
            return
        if len(self) <= max(self.p.breakout, self.p.exit_period, self.p.atr_period):
            return
        if math.isnan(self.highest_high[-1]) or math.isnan(self.lowest_low[-1]) or math.isnan(self.atr[0]):
            return

        breakout_signal = self.data.close[0] > self.highest_high[-1]
        channel_exit = self.data.close[0] < self.lowest_low[-1]

        if not self.position:
            if breakout_signal:
                self._buy_full()
            return

        self.highest_close = max(self.highest_close or float(self.data.close[0]), float(self.data.close[0]))
        atr_exit = self.data.close[0] < self.highest_close - self.p.atr_mult * self.atr[0]

        if channel_exit or atr_exit:
            self._go_flat()


class DonchianVolTargetStrategy(BaseIndexStrategy):
    """Donchian breakout with volatility-target sizing but no ATR trailing overlay."""

    params = (
        ("breakout", 55),
        ("exit_period", 30),
        ("vol_period", 30),
        ("target_vol", 0.30),
        ("max_weight", 1.0),
        ("rebalance_band", 0.15),
    )

    def __init__(self):
        super().__init__()
        self.highest_high = bt.indicators.Highest(self.data.high, period=self.p.breakout)
        self.lowest_low = bt.indicators.Lowest(self.data.low, period=self.p.exit_period)
        returns = bt.indicators.PctChange(self.data.close, period=1)
        self.volatility = bt.indicators.StdDev(returns, period=self.p.vol_period)

    def next(self):
        super().next()
        if self.order:
            return
        if len(self) <= max(self.p.breakout, self.p.exit_period, self.p.vol_period):
            return
        if (
            math.isnan(self.highest_high[-1])
            or math.isnan(self.lowest_low[-1])
            or math.isnan(self.volatility[0])
            or self.volatility[0] <= 0
        ):
            return

        breakout_signal = self.data.close[0] > self.highest_high[-1]
        exit_signal = self.data.close[0] < self.lowest_low[-1]

        if not self.position:
            if not breakout_signal:
                return
            annualized_vol = self.volatility[0] * math.sqrt(TRADING_DAYS)
            target_weight = min(self.p.max_weight, self.p.target_vol / annualized_vol)
            self._rebalance_to_weight(max(0.0, target_weight))
            return

        if exit_signal:
            self._go_flat()
            return

        annualized_vol = self.volatility[0] * math.sqrt(TRADING_DAYS)
        target_weight = min(self.p.max_weight, self.p.target_vol / annualized_vol)
        portfolio_value = self.broker.getvalue()
        current_weight = 0.0
        if portfolio_value > 0:
            current_weight = (abs(self.position.size) * self.data.close[0]) / portfolio_value
        if abs(current_weight - target_weight) >= self.p.rebalance_band:
            self._rebalance_to_weight(max(0.0, target_weight))


class DonchianHybridVolAtrStrategy(BaseIndexStrategy):
    """Donchian breakout with vol-target sizing and ATR trailing protection."""

    params = (
        ("breakout", 55),
        ("exit_period", 30),
        ("vol_period", 30),
        ("target_vol", 0.30),
        ("max_weight", 1.0),
        ("rebalance_band", 0.15),
        ("atr_period", 20),
        ("atr_mult", 4.0),
    )

    def __init__(self):
        super().__init__()
        self.highest_high = bt.indicators.Highest(self.data.high, period=self.p.breakout)
        self.lowest_low = bt.indicators.Lowest(self.data.low, period=self.p.exit_period)
        returns = bt.indicators.PctChange(self.data.close, period=1)
        self.volatility = bt.indicators.StdDev(returns, period=self.p.vol_period)
        self.atr = bt.indicators.ATR(self.data, period=self.p.atr_period)
        self.highest_close = None

    def notify_order(self, order):
        super().notify_order(order)
        if order.status != order.Completed:
            return
        if self.position:
            self.highest_close = max(self.highest_close or order.executed.price, order.executed.price)
        else:
            self.highest_close = None

    def next(self):
        super().next()
        if self.order:
            return
        if len(self) <= max(self.p.breakout, self.p.exit_period, self.p.vol_period, self.p.atr_period):
            return
        if (
            math.isnan(self.highest_high[-1])
            or math.isnan(self.lowest_low[-1])
            or math.isnan(self.volatility[0])
            or math.isnan(self.atr[0])
            or self.volatility[0] <= 0
        ):
            return

        breakout_signal = self.data.close[0] > self.highest_high[-1]
        channel_exit = self.data.close[0] < self.lowest_low[-1]

        if not self.position:
            if not breakout_signal:
                return
            annualized_vol = self.volatility[0] * math.sqrt(TRADING_DAYS)
            target_weight = min(self.p.max_weight, self.p.target_vol / annualized_vol)
            self._rebalance_to_weight(max(0.0, target_weight))
            return

        if self.highest_close is None:
            self.highest_close = self.data.close[0]
        self.highest_close = max(self.highest_close, self.data.close[0])
        trailing_stop = self.highest_close - self.p.atr_mult * self.atr[0]

        if channel_exit or self.data.close[0] < trailing_stop:
            self._go_flat()
            return

        annualized_vol = self.volatility[0] * math.sqrt(TRADING_DAYS)
        target_weight = min(self.p.max_weight, self.p.target_vol / annualized_vol)
        target_weight = max(0.0, target_weight)

        portfolio_value = self.broker.getvalue()
        current_weight = 0.0
        if portfolio_value > 0:
            current_weight = (abs(self.position.size) * self.data.close[0]) / portfolio_value

        if abs(current_weight - target_weight) >= self.p.rebalance_band:
            self._rebalance_to_weight(target_weight)


class MomentumVolTargetBasicStrategy(BaseIndexStrategy):
    """Dual-window momentum with volatility-target sizing but no regime filter."""

    params = (
        ("mom_short", 20),
        ("mom_long", 120),
        ("vol_period", 30),
        ("target_vol", 0.30),
        ("max_weight", 1.0),
        ("rebalance_band", 0.15),
    )

    def __init__(self):
        super().__init__()
        close = self.data.close
        self.roc_short = bt.indicators.ROC(close, period=self.p.mom_short)
        self.roc_long = bt.indicators.ROC(close, period=self.p.mom_long)
        returns = bt.indicators.PctChange(close, period=1)
        self.volatility = bt.indicators.StdDev(returns, period=self.p.vol_period)

    def next(self):
        super().next()
        if self.order:
            return
        if (
            math.isnan(self.roc_short[0])
            or math.isnan(self.roc_long[0])
            or math.isnan(self.volatility[0])
            or self.volatility[0] <= 0
        ):
            return

        signal = self.roc_short[0] > 0 and self.roc_long[0] > 0

        if not signal:
            self._go_flat()
            return

        annualized_vol = self.volatility[0] * math.sqrt(TRADING_DAYS)
        target_weight = min(self.p.max_weight, self.p.target_vol / annualized_vol)
        target_weight = max(0.0, target_weight)

        portfolio_value = self.broker.getvalue()
        current_weight = 0.0
        if portfolio_value > 0:
            current_weight = (abs(self.position.size) * self.data.close[0]) / portfolio_value

        if not self.position or abs(current_weight - target_weight) >= self.p.rebalance_band:
            self._rebalance_to_weight(target_weight)


class MomentumVolTargetStrategy(BaseIndexStrategy):
    """Momentum regime filter with volatility-capped position scaling."""

    params = (
        ("mom_short", 20),
        ("mom_long", 120),
        ("regime", 150),
        ("vol_period", 30),
        ("target_vol", 0.30),
        ("max_weight", 1.0),
        ("rebalance_band", 0.15),
    )

    def __init__(self):
        super().__init__()
        close = self.data.close
        self.roc_short = bt.indicators.ROC(close, period=self.p.mom_short)
        self.roc_long = bt.indicators.ROC(close, period=self.p.mom_long)
        self.sma_regime = bt.indicators.SMA(close, period=self.p.regime)
        returns = bt.indicators.PctChange(close, period=1)
        self.volatility = bt.indicators.StdDev(returns, period=self.p.vol_period)

    def next(self):
        super().next()
        if self.order:
            return
        if (
            math.isnan(self.roc_short[0])
            or math.isnan(self.roc_long[0])
            or math.isnan(self.sma_regime[0])
            or math.isnan(self.volatility[0])
            or self.volatility[0] <= 0
        ):
            return

        signal = (
            self.roc_short[0] > 0
            and self.roc_long[0] > 0
            and self.data.close[0] > self.sma_regime[0]
        )

        if not signal:
            self._go_flat()
            return

        annualized_vol = self.volatility[0] * math.sqrt(TRADING_DAYS)
        target_weight = min(self.p.max_weight, self.p.target_vol / annualized_vol)
        target_weight = max(0.0, target_weight)

        portfolio_value = self.broker.getvalue()
        current_weight = 0.0
        if portfolio_value > 0:
            current_weight = (abs(self.position.size) * self.data.close[0]) / portfolio_value

        if not self.position or abs(current_weight - target_weight) >= self.p.rebalance_band:
            self._rebalance_to_weight(target_weight)


class SuperTrendLongMaStrategy(BaseIndexStrategy):
    """SuperTrend breakout gated by a longer moving-average regime."""

    params = (
        ("supertrend_period", 10),
        ("supertrend_mult", 3.0),
        ("regime", 200),
    )

    def __init__(self):
        super().__init__()
        self.supertrend = SuperTrendIndicator(
            self.data,
            period=self.p.supertrend_period,
            multiplier=self.p.supertrend_mult,
        )
        self.sma_regime = bt.indicators.SMA(self.data.close, period=self.p.regime)

    def next(self):
        super().next()
        if self.order:
            return
        if math.isnan(self.supertrend[0]) or math.isnan(self.sma_regime[0]):
            return

        signal = self.data.close[0] > self.supertrend[0] and self.data.close[0] > self.sma_regime[0]

        if signal and not self.position:
            self._buy_full()
        elif self.position and not signal:
            self._go_flat()


class SuperTrendBasicStrategy(BaseIndexStrategy):
    """Pure SuperTrend breakout without the long-MA regime filter."""

    params = (
        ("supertrend_period", 10),
        ("supertrend_mult", 3.0),
    )

    def __init__(self):
        super().__init__()
        self.supertrend = SuperTrendIndicator(
            self.data,
            period=self.p.supertrend_period,
            multiplier=self.p.supertrend_mult,
        )

    def next(self):
        super().next()
        if self.order:
            return
        if math.isnan(self.supertrend[0]):
            return

        signal = self.data.close[0] > self.supertrend[0]

        if signal and not self.position:
            self._buy_full()
        elif self.position and not signal:
            self._go_flat()


class TsmomLooseThresholdStrategy(BaseIndexStrategy):
    """Multi-window time-series momentum with regime filter but a looser positivity threshold."""

    params = (
        ("mom_windows", (60, 120, 240)),
        ("positive_threshold", 1),
        ("regime", 200),
    )

    def __init__(self):
        super().__init__()
        close = self.data.close
        self.momentum = [bt.indicators.ROC(close, period=window) for window in self.p.mom_windows]
        self.sma_regime = bt.indicators.SMA(close, period=self.p.regime)

    def next(self):
        super().next()
        if self.order:
            return
        if math.isnan(self.sma_regime[0]) or any(math.isnan(indicator[0]) for indicator in self.momentum):
            return

        positive_count = sum(indicator[0] > 0 for indicator in self.momentum)
        signal = positive_count >= self.p.positive_threshold and self.data.close[0] > self.sma_regime[0]

        if signal and not self.position:
            self._buy_full()
        elif self.position and not signal:
            self._go_flat()


class TsmomBasicStrategy(BaseIndexStrategy):
    """Multi-window time-series momentum without the long-MA regime filter."""

    params = (
        ("mom_windows", (60, 120, 240)),
        ("positive_threshold", 2),
    )

    def __init__(self):
        super().__init__()
        close = self.data.close
        self.momentum = [bt.indicators.ROC(close, period=window) for window in self.p.mom_windows]

    def next(self):
        super().next()
        if self.order:
            return
        if any(math.isnan(indicator[0]) for indicator in self.momentum):
            return

        positive_count = sum(indicator[0] > 0 for indicator in self.momentum)
        signal = positive_count >= self.p.positive_threshold

        if signal and not self.position:
            self._buy_full()
        elif self.position and not signal:
            self._go_flat()


class TsmomRegimeStrategy(BaseIndexStrategy):
    """Multi-window time-series momentum with a long MA regime filter."""

    params = (
        ("mom_windows", (60, 120, 240)),
        ("positive_threshold", 2),
        ("regime", 200),
    )

    def __init__(self):
        super().__init__()
        close = self.data.close
        self.momentum = [bt.indicators.ROC(close, period=window) for window in self.p.mom_windows]
        self.sma_regime = bt.indicators.SMA(close, period=self.p.regime)

    def next(self):
        super().next()
        if self.order:
            return
        if math.isnan(self.sma_regime[0]) or any(math.isnan(indicator[0]) for indicator in self.momentum):
            return

        positive_count = sum(indicator[0] > 0 for indicator in self.momentum)
        signal = positive_count >= self.p.positive_threshold and self.data.close[0] > self.sma_regime[0]

        if signal and not self.position:
            self._buy_full()
        elif self.position and not signal:
            self._go_flat()


class KamaBasicStrategy(BaseIndexStrategy):
    """KAMA trend following without the long-MA regime filter."""

    params = (
        ("kama_period", 30),
        ("kama_fast", 2),
        ("kama_slow", 30),
    )

    def __init__(self):
        super().__init__()
        close = self.data.close
        self.kama = bt.indicators.KAMA(
            close,
            period=self.p.kama_period,
            fast=self.p.kama_fast,
            slow=self.p.kama_slow,
        )

    def next(self):
        super().next()
        if self.order:
            return
        if len(self) <= self.p.kama_period:
            return
        if math.isnan(self.kama[0]) or math.isnan(self.kama[-1]):
            return

        signal = self.data.close[0] > self.kama[0] and self.kama[0] > self.kama[-1]

        if signal and not self.position:
            self._buy_full()
        elif self.position and not signal:
            self._go_flat()


class KamaTrendStrategy(BaseIndexStrategy):
    """KAMA trend following with slope confirmation and long MA regime."""

    params = (
        ("kama_period", 30),
        ("kama_fast", 2),
        ("kama_slow", 30),
        ("regime", 200),
    )

    def __init__(self):
        super().__init__()
        close = self.data.close
        self.kama = bt.indicators.KAMA(
            close,
            period=self.p.kama_period,
            fast=self.p.kama_fast,
            slow=self.p.kama_slow,
        )
        self.sma_regime = bt.indicators.SMA(close, period=self.p.regime)

    def next(self):
        super().next()
        if self.order:
            return
        if len(self) <= max(self.p.kama_period, self.p.regime):
            return
        if math.isnan(self.kama[0]) or math.isnan(self.kama[-1]) or math.isnan(self.sma_regime[0]):
            return

        signal = (
            self.data.close[0] > self.kama[0]
            and self.kama[0] > self.kama[-1]
            and self.data.close[0] > self.sma_regime[0]
        )

        if signal and not self.position:
            self._buy_full()
        elif self.position and not signal:
            self._go_flat()


class DualThrustRegimeStrategy(BaseIndexStrategy):
    """Close-only Dual Thrust proxy for daily index data with no intraday range."""

    params = (
        ("range_period", 20),
        ("k1", 0.5),
        ("k2", 0.5),
        ("regime", 200),
    )

    def __init__(self):
        super().__init__()
        self.sma_regime = bt.indicators.SMA(self.data.close, period=self.p.regime)

    def next(self):
        super().next()
        if self.order:
            return
        if len(self) <= max(self.p.range_period, self.p.regime):
            return
        if math.isnan(self.sma_regime[0]):
            return

        closes = [float(self.data.close[-offset]) for offset in range(1, self.p.range_period + 1)]
        thrust_range = max(closes) - min(closes)
        reference_price = float(self.data.close[-1])
        upper = reference_price + self.p.k1 * thrust_range
        lower = reference_price - self.p.k2 * thrust_range

        entry_signal = self.data.close[0] > upper and self.data.close[0] > self.sma_regime[0]
        exit_signal = self.data.close[0] < lower or self.data.close[0] < self.sma_regime[0]

        if entry_signal and not self.position:
            self._buy_full()
        elif self.position and exit_signal:
            self._go_flat()


class DualThrustBasicStrategy(BaseIndexStrategy):
    """Close-only Dual Thrust proxy without regime filter."""

    params = (
        ("range_period", 20),
        ("k1", 0.5),
        ("k2", 0.5),
    )

    def __init__(self):
        super().__init__()

    def next(self):
        super().next()
        if self.order:
            return
        if len(self) <= self.p.range_period:
            return

        closes = [float(self.data.close[-offset]) for offset in range(1, self.p.range_period + 1)]
        thrust_range = max(closes) - min(closes)
        reference_price = float(self.data.close[-1])
        upper = reference_price + self.p.k1 * thrust_range
        lower = reference_price - self.p.k2 * thrust_range

        entry_signal = self.data.close[0] > upper
        exit_signal = self.data.close[0] < lower

        if entry_signal and not self.position:
            self._buy_full()
        elif self.position and exit_signal:
            self._go_flat()


class MacdBasicStrategy(BaseIndexStrategy):
    """MACD trend timing without the long moving-average regime filter."""

    params = (
        ("macd_fast", 12),
        ("macd_slow", 26),
        ("macd_signal", 9),
    )

    def __init__(self):
        super().__init__()
        close = self.data.close
        self.macd = bt.indicators.MACD(
            close,
            period_me1=self.p.macd_fast,
            period_me2=self.p.macd_slow,
            period_signal=self.p.macd_signal,
        )

    def next(self):
        super().next()
        if self.order:
            return
        if math.isnan(self.macd.macd[0]) or math.isnan(self.macd.signal[0]):
            return

        signal = self.macd.macd[0] > self.macd.signal[0]

        if signal and not self.position:
            self._buy_full()
        elif self.position and not signal:
            self._go_flat()


class MacdLongMaStrategy(BaseIndexStrategy):
    """MACD trend timing with a longer moving-average regime filter."""

    params = (
        ("macd_fast", 12),
        ("macd_slow", 26),
        ("macd_signal", 9),
        ("regime", 200),
    )

    def __init__(self):
        super().__init__()
        close = self.data.close
        self.macd = bt.indicators.MACD(
            close,
            period_me1=self.p.macd_fast,
            period_me2=self.p.macd_slow,
            period_signal=self.p.macd_signal,
        )
        self.sma_regime = bt.indicators.SMA(close, period=self.p.regime)

    def next(self):
        super().next()
        if self.order:
            return
        if math.isnan(self.macd.macd[0]) or math.isnan(self.macd.signal[0]) or math.isnan(self.sma_regime[0]):
            return

        signal = self.macd.macd[0] > self.macd.signal[0] and self.data.close[0] > self.sma_regime[0]

        if signal and not self.position:
            self._buy_full()
        elif self.position and not signal:
            self._go_flat()


def load_dataframe() -> pd.DataFrame:
    df = pd.read_csv(DATA_FILE, parse_dates=["datetime"], index_col="datetime")
    df = df.sort_index()
    return df[["open", "high", "low", "close", "volume"]].copy()


def run_strategy(strategy_cls, config: dict, df: pd.DataFrame | None = None) -> dict:
    if df is None:
        df = load_dataframe()
    cerebro = bt.Cerebro(stdstats=False)
    cerebro.adddata(Chinext50Data(dataname=df))
    cerebro.addstrategy(strategy_cls, **config)
    cerebro.broker.setcash(INITIAL_CASH)
    cerebro.broker.setcommission(commission=COMMISSION)
    cerebro.addanalyzer(bt.analyzers.Returns, _name="returns")
    cerebro.addanalyzer(bt.analyzers.DrawDown, _name="drawdown")
    cerebro.addanalyzer(bt.analyzers.SharpeRatio_A, _name="sharpe", riskfreerate=0.02)
    cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name="trades")

    strategy = cerebro.run()[0]
    final_value = cerebro.broker.getvalue()
    returns = strategy.analyzers.returns.get_analysis()
    drawdown = strategy.analyzers.drawdown.get_analysis()
    sharpe = strategy.analyzers.sharpe.get_analysis()
    trades = strategy.analyzers.trades.get_analysis()
    closed_trades = trades.get("total", {}).get("closed", 0)
    won_trades = trades.get("won", {}).get("total", 0)
    total_bars = len(df)

    return {
        "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),
        "max_drawdown_pct": round(drawdown.get("max", {}).get("drawdown", 0.0), 2),
        "sharpe": round(sharpe["sharperatio"], 3) if sharpe.get("sharperatio") is not None else None,
        "entries": strategy.entry_count,
        "closed_trades": closed_trades,
        "win_rate_pct": round((won_trades / closed_trades) * 100.0, 2) if closed_trades else 0.0,
        "exposure_pct": round((strategy.exposure_sum / total_bars) * 100.0, 2),
    }


def format_config(config: dict) -> str:
    return ", ".join(f"`{key}={value}`" for key, value in config.items())


def format_config_plain(config: dict) -> str:
    return ", ".join(f"{key}={value}" for key, value in config.items())


def metric_text(value):
    return "N/A" if value is None else value


def sharpe_sort_value(item: dict) -> float:
    sharpe = item["metrics"]["sharpe"]
    return float("-inf") if sharpe is None else sharpe


def choose_tradeoff_winner(results: list[dict]) -> dict:
    return max(
        results,
        key=lambda item: (
            sharpe_sort_value(item),
            item["metrics"]["annual_return_pct"],
            -item["metrics"]["max_drawdown_pct"],
        ),
    )


def choose_capped_drawdown_winner(results: list[dict]) -> tuple[dict, float | None]:
    for drawdown_cap in (35.0, 40.0):
        eligible = [
            item
            for item in results
            if item["metrics"]["sharpe"] is not None and item["metrics"]["max_drawdown_pct"] <= drawdown_cap
        ]
        if eligible:
            return choose_tradeoff_winner(eligible), drawdown_cap

    eligible = [item for item in results if item["metrics"]["sharpe"] is not None]
    if eligible:
        return choose_tradeoff_winner(eligible), None
    return min(results, key=lambda item: item["metrics"]["max_drawdown_pct"]), None


def choose_best_by_drawdown_cap(results: list[dict], drawdown_cap: float) -> dict | None:
    eligible = [item for item in results if item["metrics"]["max_drawdown_pct"] <= drawdown_cap]
    if not eligible:
        return None
    return choose_tradeoff_winner(eligible)


def dualthrust_search_configs() -> list[dict]:
    range_periods = [10, 15, 20, 25, 30]
    k_values = [0.2, 0.3, 0.4, 0.5]
    regimes = [60, 90, 120, 150, 200]
    configs = []
    for range_period, k1, k2, regime in itertools.product(range_periods, k_values, k_values, regimes):
        configs.append(
            {
                "range_period": range_period,
                "k1": round(k1, 1),
                "k2": round(k2, 1),
                "regime": regime,
            }
        )
    return configs


def summarize_dualthrust_candidate(label: str, item: dict | None) -> list[str]:
    if item is None:
        return [f"- {label}: no configuration met the filter."]
    metrics = item["metrics"]
    return [
        (
            f"- {label}: `{format_config_plain(item['config'])}` "
            f"| annual return `{metrics['annual_return_pct']:.2f}%` "
            f"| Sharpe `{metric_text(metrics['sharpe'])}` "
            f"| max DD `{metrics['max_drawdown_pct']:.2f}%` "
            f"| trades `{metrics['closed_trades']}`"
        )
    ]


def build_dualthrust_optimization_report(results: list[dict], default_result: dict, df: pd.DataFrame) -> str:
    best_return = max(results, key=lambda item: item["metrics"]["annual_return_pct"])
    best_sharpe = choose_tradeoff_winner(results)
    best_dd30 = choose_best_by_drawdown_cap(results, 30.0)
    best_dd35 = choose_best_by_drawdown_cap(results, 35.0)
    top_by_return = sorted(
        results,
        key=lambda item: (
            item["metrics"]["annual_return_pct"],
            sharpe_sort_value(item),
            -item["metrics"]["max_drawdown_pct"],
        ),
        reverse=True,
    )[:8]
    top_by_sharpe = sorted(
        results,
        key=lambda item: (
            sharpe_sort_value(item),
            item["metrics"]["annual_return_pct"],
            -item["metrics"]["max_drawdown_pct"],
        ),
        reverse=True,
    )[:8]

    recommended = best_dd35 or best_sharpe
    default_metrics = default_result["metrics"]
    recommended_metrics = recommended["metrics"]
    beats_default_return = best_return["metrics"]["annual_return_pct"] > default_metrics["annual_return_pct"]
    beats_default_sharpe = (
        best_sharpe["metrics"]["sharpe"] is not None
        and (
            default_metrics["sharpe"] is None
            or best_sharpe["metrics"]["sharpe"] > default_metrics["sharpe"]
        )
    )

    lines = [
        "# DualThrust Chinext50 Optimization",
        "",
        f"- Data: `chinext50.csv` ({df.index.min().date()} to {df.index.max().date()}, {len(df)} bars)",
        f"- Initial cash: `{INITIAL_CASH:.0f}`",
        f"- Commission: `{COMMISSION:.4f}`",
        "- Strategy: `DualThrustRegimeStrategy`",
        "- Search grid: `range_period in [10, 15, 20, 25, 30]`, `k1/k2 in [0.2, 0.3, 0.4, 0.5]`, `regime in [60, 90, 120, 150, 200]`",
        f"- Evaluated configs: `{len(results)}`",
        "",
        "## Command",
        "",
        "- `python3 chinext50_experiments.py --optimize-dualthrust`",
        "",
        "## Default Benchmark",
        "",
        (
            f"- Current default: `{format_config_plain(default_result['config'])}` "
            f"| annual return `{default_metrics['annual_return_pct']:.2f}%` "
            f"| Sharpe `{metric_text(default_metrics['sharpe'])}` "
            f"| max DD `{default_metrics['max_drawdown_pct']:.2f}%`"
        ),
        "",
        "## Required Winners",
        "",
    ]

    lines.extend(summarize_dualthrust_candidate("Best by annual return", best_return))
    lines.extend(summarize_dualthrust_candidate("Best by Sharpe", best_sharpe))
    lines.extend(summarize_dualthrust_candidate("Best with max DD <= 30% (Sharpe-ranked)", best_dd30))
    lines.extend(summarize_dualthrust_candidate("Best with max DD <= 35% (Sharpe-ranked)", best_dd35))

    lines.extend(
        [
            "",
            "## Top Candidates By Annual Return",
            "",
            "| Params | Annual Return | Sharpe | Max DD | Trades | Exposure |",
            "| --- | ---: | ---: | ---: | ---: | ---: |",
        ]
    )

    for item in top_by_return:
        metrics = item["metrics"]
        lines.append(
            "| {params} | {annual_return_pct:.2f}% | {sharpe} | {max_drawdown_pct:.2f}% | {closed_trades} | {exposure_pct:.2f}% |".format(
                params=format_config(item["config"]),
                annual_return_pct=metrics["annual_return_pct"],
                sharpe=metric_text(metrics["sharpe"]),
                max_drawdown_pct=metrics["max_drawdown_pct"],
                closed_trades=metrics["closed_trades"],
                exposure_pct=metrics["exposure_pct"],
            )
        )

    lines.extend(
        [
            "",
            "## Top Candidates By Sharpe",
            "",
            "| Params | Sharpe | Annual Return | Max DD | Trades | Exposure |",
            "| --- | ---: | ---: | ---: | ---: | ---: |",
        ]
    )

    for item in top_by_sharpe:
        metrics = item["metrics"]
        lines.append(
            "| {params} | {sharpe} | {annual_return_pct:.2f}% | {max_drawdown_pct:.2f}% | {closed_trades} | {exposure_pct:.2f}% |".format(
                params=format_config(item["config"]),
                sharpe=metric_text(metrics["sharpe"]),
                annual_return_pct=metrics["annual_return_pct"],
                max_drawdown_pct=metrics["max_drawdown_pct"],
                closed_trades=metrics["closed_trades"],
                exposure_pct=metrics["exposure_pct"],
            )
        )

    lines.extend(
        [
            "",
            "## Recommendation",
            "",
            (
                f"- Recommended next parameter set: `{format_config_plain(recommended['config'])}` "
                f"because it keeps max DD at `{recommended_metrics['max_drawdown_pct']:.2f}%` while delivering "
                f"`{recommended_metrics['annual_return_pct']:.2f}%` annual return and Sharpe `{metric_text(recommended_metrics['sharpe'])}`."
            ),
            (
                f"- Optimized DualThrust beat default on annual return: `{'yes' if beats_default_return else 'no'}` "
                f"({best_return['metrics']['annual_return_pct']:.2f}% vs {default_metrics['annual_return_pct']:.2f}%)."
            ),
            (
                f"- Optimized DualThrust beat default on Sharpe: `{'yes' if beats_default_sharpe else 'no'}` "
                f"({metric_text(best_sharpe['metrics']['sharpe'])} vs {metric_text(default_metrics['sharpe'])})."
            ),
        ]
    )

    return "\n".join(lines) + "\n"


def build_report(results: list[dict], df: pd.DataFrame) -> str:
    best_return = max(results, key=lambda item: item["metrics"]["annual_return_pct"])
    best_sharpe = choose_tradeoff_winner(results)
    best_capped_drawdown, drawdown_cap = choose_capped_drawdown_winner(results)
    previous_results = [item for item in results if not item.get("is_new")]
    new_results = [item for item in results if item.get("is_new")]
    previous_best_return = max(previous_results, key=lambda item: item["metrics"]["annual_return_pct"]) if previous_results else None
    previous_best_sharpe = choose_tradeoff_winner(previous_results) if previous_results else None
    new_best_return = max(new_results, key=lambda item: item["metrics"]["annual_return_pct"]) if new_results else None
    new_best_sharpe = choose_tradeoff_winner(new_results) if new_results else None
    lines = [
        "# Chinext50 Backtrader Experiments",
        "",
        f"- Data: `chinext50.csv` ({df.index.min().date()} to {df.index.max().date()}, {len(df)} bars)",
        f"- Initial cash: `{INITIAL_CASH:.0f}`",
        f"- Commission: `{COMMISSION:.4f}`",
        "",
        "## Commands",
        "",
        "- Run all experiments: `python3 chinext50_experiments.py`",
        "",
        "## Configs",
        "",
    ]

    for item in results:
        lines.append(f"- **{item['name']}**: {format_config(item['config'])}")

    lines.extend(
        [
            "",
            "## Metrics",
            "",
            "| Strategy | Final Value | Total Return | Annual Return | Sharpe | Max DD | Entries | Closed Trades | Win Rate | Avg Exposure |",
            "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |",
        ]
    )

    for item in results:
        metrics = item["metrics"]
        lines.append(
            "| {name} | {final_value:.2f} | {total_return_pct:.2f}% | {annual_return_pct:.2f}% | {sharpe} | {max_drawdown_pct:.2f}% | {entries} | {closed_trades} | {win_rate_pct:.2f}% | {exposure_pct:.2f}% |".format(
                name=item["name"],
                final_value=metrics["final_value"],
                total_return_pct=metrics["total_return_pct"],
                annual_return_pct=metrics["annual_return_pct"],
                sharpe=metric_text(metrics["sharpe"]),
                max_drawdown_pct=metrics["max_drawdown_pct"],
                entries=metrics["entries"],
                closed_trades=metrics["closed_trades"],
                win_rate_pct=metrics["win_rate_pct"],
                exposure_pct=metrics["exposure_pct"],
            )
        )

    lines.extend(
        [
            "",
            "## Verdict",
            "",
            (
                f"- Highest annual return: **{best_return['name']}** "
                f"({best_return['metrics']['annual_return_pct']:.2f}% annual return, "
                f"{best_return['metrics']['max_drawdown_pct']:.2f}% max DD)"
            ),
            (
                f"- Best risk-adjusted balance by Sharpe: **{best_sharpe['name']}** "
                f"(Sharpe {metric_text(best_sharpe['metrics']['sharpe'])}, "
                f"{best_sharpe['metrics']['annual_return_pct']:.2f}% annual return, "
                f"{best_sharpe['metrics']['max_drawdown_pct']:.2f}% max DD)"
            ),
        ]
    )
    if drawdown_cap is not None:
        lines.append(
            f"- Best Sharpe with max DD <= {drawdown_cap:.0f}%: **{best_capped_drawdown['name']}** "
            f"(Sharpe {metric_text(best_capped_drawdown['metrics']['sharpe'])}, "
            f"{best_capped_drawdown['metrics']['annual_return_pct']:.2f}% annual return, "
            f"{best_capped_drawdown['metrics']['max_drawdown_pct']:.2f}% max DD)"
        )
    else:
        lines.append(
            f"- Best drawdown-aware fallback: **{best_capped_drawdown['name']}** "
            f"(Sharpe {metric_text(best_capped_drawdown['metrics']['sharpe'])}, "
            f"{best_capped_drawdown['metrics']['annual_return_pct']:.2f}% annual return, "
            f"{best_capped_drawdown['metrics']['max_drawdown_pct']:.2f}% max DD)"
        )

    if previous_best_return and new_best_return:
        if new_best_return["metrics"]["annual_return_pct"] > previous_best_return["metrics"]["annual_return_pct"]:
            lines.append(
                f"- New-strategy return leader: **{new_best_return['name']}** beat the prior return leader "
                f"**{previous_best_return['name']}** "
                f"({new_best_return['metrics']['annual_return_pct']:.2f}% vs {previous_best_return['metrics']['annual_return_pct']:.2f}% annual return)"
            )
        else:
            lines.append(
                f"- New strategies did not beat the prior return leader **{previous_best_return['name']}** "
                f"({previous_best_return['metrics']['annual_return_pct']:.2f}% annual return). "
                f"Best new result was **{new_best_return['name']}** at {new_best_return['metrics']['annual_return_pct']:.2f}%."
            )

    if previous_best_sharpe and new_best_sharpe:
        previous_sharpe = previous_best_sharpe["metrics"]["sharpe"]
        new_sharpe = new_best_sharpe["metrics"]["sharpe"]
        if new_sharpe is not None and (previous_sharpe is None or new_sharpe > previous_sharpe):
            lines.append(
                f"- New-strategy Sharpe leader: **{new_best_sharpe['name']}** beat the prior Sharpe leader "
                f"**{previous_best_sharpe['name']}** "
                f"(Sharpe {metric_text(new_sharpe)} vs {metric_text(previous_sharpe)})"
            )
        else:
            lines.append(
                f"- New strategies did not beat the prior Sharpe leader **{previous_best_sharpe['name']}** "
                f"(Sharpe {metric_text(previous_sharpe)}). "
                f"Best new Sharpe was **{new_best_sharpe['name']}** at {metric_text(new_sharpe)}."
            )

    if new_results:
        best_new_drawdown = min(new_results, key=lambda item: item["metrics"]["max_drawdown_pct"])
        lines.append(
            f"- Most defensive new addition: **{best_new_drawdown['name']}** "
            f"delivered the lowest max DD among the new strategies "
            f"with {best_new_drawdown['metrics']['max_drawdown_pct']:.2f}% max DD, "
            f"{best_new_drawdown['metrics']['annual_return_pct']:.2f}% annual return, "
            f"and {best_new_drawdown['metrics']['win_rate_pct']:.2f}% win rate."
        )

    return "\n".join(lines) + "\n"


def run_dualthrust_optimization() -> list[dict]:
    df = load_dataframe()
    configs = dualthrust_search_configs()
    results = []

    for config in configs:
        metrics = run_strategy(DualThrustRegimeStrategy, config, df=df)
        results.append(
            {
                "name": "DualThrustRegimeStrategy",
                "config": config,
                "metrics": metrics,
            }
        )

    default_config = {"range_period": 20, "k1": 0.3, "k2": 0.3, "regime": 120}
    default_result = next(item for item in results if item["config"] == default_config)
    report = build_dualthrust_optimization_report(results, default_result, df)
    DUALTHRUST_OPT_REPORT_FILE.write_text(report, encoding="utf-8")

    best_return = max(results, key=lambda item: item["metrics"]["annual_return_pct"])
    best_sharpe = choose_tradeoff_winner(results)
    best_dd30 = choose_best_by_drawdown_cap(results, 30.0)
    best_dd35 = choose_best_by_drawdown_cap(results, 35.0)

    print(f"Evaluated {len(results)} DualThrust configurations")
    for label, item in (
        ("Default", default_result),
        ("Best annual return", best_return),
        ("Best Sharpe", best_sharpe),
        ("Best max DD <= 30%", best_dd30),
        ("Best max DD <= 35%", best_dd35),
    ):
        if item is None:
            print(f"{label}: none")
            continue
        metrics = item["metrics"]
        print(
            f"{label}: config={item['config']}, "
            f"annual_return={metrics['annual_return_pct']:.2f}%, "
            f"sharpe={metric_text(metrics['sharpe'])}, "
            f"max_dd={metrics['max_drawdown_pct']:.2f}%"
        )
    print(f"Report written to {DUALTHRUST_OPT_REPORT_FILE.name}")
    return results


def run_experiments() -> list[dict]:
    df = load_dataframe()
    experiments = [
        {
            "name": "TrendRegimeFlatStrategy",
            "strategy": TrendRegimeFlatStrategy,
            "config": {"fast": 20, "slow": 60, "regime": 120, "vol_fast": 20, "vol_slow": 60, "vol_cap": 1.10},
        },
        {
            "name": "TrendTightVolStrategy",
            "strategy": TrendTightVolStrategy,
            "config": {"fast": 20, "slow": 60, "regime": 120, "vol_fast": 20, "vol_slow": 60, "vol_cap": 0.95},
            "is_new": True,
        },
        {
            "name": "TrendLooseVolStrategy",
            "strategy": TrendLooseVolStrategy,
            "config": {"fast": 20, "slow": 60, "regime": 120, "vol_fast": 20, "vol_slow": 60, "vol_cap": 1.25},
            "is_new": True,
        },
        {
            "name": "SmaLongFilterTrendStrategy",
            "strategy": SmaLongFilterTrendStrategy,
            "config": {"fast": 20, "slow": 60, "regime": 120},
            "is_new": True,
        },
        {
            "name": "MomentumBasicStrategy",
            "strategy": MomentumBasicStrategy,
            "config": {"mom_short": 20, "mom_long": 120},
            "is_new": True,
        },
        {
            "name": "MomentumVolTargetBasicStrategy",
            "strategy": MomentumVolTargetBasicStrategy,
            "config": {"mom_short": 20, "mom_long": 120, "vol_period": 30, "target_vol": 0.30, "max_weight": 1.0, "rebalance_band": 0.15},
            "is_new": True,
        },
        {
            "name": "MomentumVolTargetLowerTargetStrategy",
            "strategy": MomentumVolTargetStrategy,
            "config": {"mom_short": 20, "mom_long": 120, "regime": 150, "vol_period": 30, "target_vol": 0.25, "max_weight": 1.0, "rebalance_band": 0.15},
            "is_new": True,
        },
        {
            "name": "MomentumVolTargetHigherTargetStrategy",
            "strategy": MomentumVolTargetStrategy,
            "config": {"mom_short": 20, "mom_long": 120, "regime": 150, "vol_period": 30, "target_vol": 0.35, "max_weight": 1.0, "rebalance_band": 0.15},
            "is_new": True,
        },
        {
            "name": "MomentumRegimeStrategy",
            "strategy": MomentumRegimeStrategy,
            "config": {"mom_short": 20, "mom_long": 120, "regime": 150},
        },
        {
            "name": "MomentumMaFilterStrategy",
            "strategy": MomentumMaFilterStrategy,
            "config": {"mom_short": 20, "mom_long": 120, "ma_filter": 60},
            "is_new": True,
        },
        {
            "name": "MomentumAtrTrailBasicStrategy",
            "strategy": MomentumAtrTrailBasicStrategy,
            "config": {"mom_short": 20, "mom_long": 120, "atr_period": 20, "atr_mult": 4.0},
            "is_new": True,
        },
        {
            "name": "MomentumAtrTrailTighterStrategy",
            "strategy": MomentumAtrTrailStrategy,
            "config": {"mom_short": 20, "mom_long": 120, "regime": 150, "atr_period": 20, "atr_mult": 3.0},
            "is_new": True,
        },
        {
            "name": "MomentumAtrTrailLooserStrategy",
            "strategy": MomentumAtrTrailStrategy,
            "config": {"mom_short": 20, "mom_long": 120, "regime": 150, "atr_period": 20, "atr_mult": 5.0},
            "is_new": True,
        },
        {
            "name": "MomentumAtrTrailStrategy",
            "strategy": MomentumAtrTrailStrategy,
            "config": {"mom_short": 20, "mom_long": 120, "regime": 150, "atr_period": 20, "atr_mult": 4.0},
        },
        {
            "name": "MomentumDefensiveFilterStrategy",
            "strategy": MomentumDefensiveFilterStrategy,
            "config": {
                "mom_short": 20,
                "mom_long": 120,
                "regime": 150,
                "vol_fast": 20,
                "vol_slow": 60,
                "vol_cap": 1.05,
            },
            "is_new": True,
        },
        {
            "name": "DonchianRegimeStrategy",
            "strategy": DonchianRegimeStrategy,
            "config": {"breakout": 55, "exit_period": 30, "regime": 150},
        },
        {
            "name": "DonchianRegimeFastStrategy",
            "strategy": DonchianRegimeStrategy,
            "config": {"breakout": 40, "exit_period": 20, "regime": 150},
            "is_new": True,
        },
        {
            "name": "DonchianRegimeSlowStrategy",
            "strategy": DonchianRegimeStrategy,
            "config": {"breakout": 70, "exit_period": 35, "regime": 150},
            "is_new": True,
        },
        {
            "name": "MomentumVolTargetStrategy",
            "strategy": MomentumVolTargetStrategy,
            "config": {
                "mom_short": 20,
                "mom_long": 120,
                "regime": 150,
                "vol_period": 30,
                "target_vol": 0.30,
                "max_weight": 1.0,
                "rebalance_band": 0.15,
            },
        },
        {
            "name": "DonchianBasicStrategy",
            "strategy": DonchianBasicStrategy,
            "config": {"breakout": 55, "exit_period": 30},
            "is_new": True,
        },
        {
            "name": "DonchianAdxStrategy",
            "strategy": DonchianAdxStrategy,
            "config": {"breakout": 55, "exit_period": 30, "adx_period": 14, "adx_threshold": 20.0},
            "is_new": True,
        },
        {
            "name": "DonchianAtrTrailStrategy",
            "strategy": DonchianAtrTrailStrategy,
            "config": {"breakout": 55, "exit_period": 30, "atr_period": 20, "atr_mult": 4.0},
            "is_new": True,
        },
        {
            "name": "DonchianVolTargetStrategy",
            "strategy": DonchianVolTargetStrategy,
            "config": {"breakout": 55, "exit_period": 30, "vol_period": 30, "target_vol": 0.30, "max_weight": 1.0, "rebalance_band": 0.15},
            "is_new": True,
        },
        {
            "name": "DonchianHybridVolAtrStrategy",
            "strategy": DonchianHybridVolAtrStrategy,
            "config": {
                "breakout": 55,
                "exit_period": 30,
                "vol_period": 30,
                "target_vol": 0.30,
                "max_weight": 1.0,
                "rebalance_band": 0.15,
                "atr_period": 20,
                "atr_mult": 4.0,
            },
            "is_new": True,
        },
        {
            "name": "SuperTrendLongMaStrategy",
            "strategy": SuperTrendLongMaStrategy,
            "config": {"supertrend_period": 14, "supertrend_mult": 2.0, "regime": 200},
            "is_new": True,
        },
        {
            "name": "SuperTrendLongMaFastRegimeStrategy",
            "strategy": SuperTrendLongMaStrategy,
            "config": {"supertrend_period": 14, "supertrend_mult": 2.0, "regime": 150},
            "is_new": True,
        },
        {
            "name": "SuperTrendBasicStrategy",
            "strategy": SuperTrendBasicStrategy,
            "config": {"supertrend_period": 14, "supertrend_mult": 2.0},
            "is_new": True,
        },
        {
            "name": "TsmomLooseThresholdStrategy",
            "strategy": TsmomLooseThresholdStrategy,
            "config": {"mom_windows": (60, 120, 240), "positive_threshold": 1, "regime": 200},
            "is_new": True,
        },
        {
            "name": "TsmomBasicStrategy",
            "strategy": TsmomBasicStrategy,
            "config": {"mom_windows": (60, 120), "positive_threshold": 1},
            "is_new": True,
        },
        {
            "name": "TsmomRegimeStrategy",
            "strategy": TsmomRegimeStrategy,
            "config": {"mom_windows": (60, 120), "positive_threshold": 1, "regime": 200},
            "is_new": True,
        },
        {
            "name": "KamaBasicStrategy",
            "strategy": KamaBasicStrategy,
            "config": {"kama_period": 30, "kama_fast": 2, "kama_slow": 30},
            "is_new": True,
        },
        {
            "name": "KamaTrendStrategy",
            "strategy": KamaTrendStrategy,
            "config": {"kama_period": 30, "kama_fast": 2, "kama_slow": 30, "regime": 120},
            "is_new": True,
        },
        {
            "name": "DualThrustBasicStrategy",
            "strategy": DualThrustBasicStrategy,
            "config": {"range_period": 20, "k1": 0.3, "k2": 0.3},
            "is_new": True,
        },
        {
            "name": "DualThrustFastStrategy",
            "strategy": DualThrustBasicStrategy,
            "config": {"range_period": 15, "k1": 0.3, "k2": 0.3},
            "is_new": True,
        },
        {
            "name": "DualThrustSlowStrategy",
            "strategy": DualThrustBasicStrategy,
            "config": {"range_period": 30, "k1": 0.3, "k2": 0.3},
            "is_new": True,
        },
        {
            "name": "DualThrustRegimeStrategy",
            "strategy": DualThrustRegimeStrategy,
            "config": {"range_period": 20, "k1": 0.3, "k2": 0.3, "regime": 120},
            "is_new": True,
        },
        {
            "name": "MacdBasicStrategy",
            "strategy": MacdBasicStrategy,
            "config": {"macd_fast": 12, "macd_slow": 26, "macd_signal": 9},
            "is_new": True,
        },
        {
            "name": "MacdLongMaFastRegimeStrategy",
            "strategy": MacdLongMaStrategy,
            "config": {"macd_fast": 12, "macd_slow": 26, "macd_signal": 9, "regime": 150},
            "is_new": True,
        },
        {
            "name": "MacdLongMaStrategy",
            "strategy": MacdLongMaStrategy,
            "config": {"macd_fast": 12, "macd_slow": 26, "macd_signal": 9, "regime": 120},
            "is_new": True,
        },
    ]

    results = []
    for item in experiments:
        metrics = run_strategy(item["strategy"], item["config"], df=df)
        results.append(
            {
                "name": item["name"],
                "config": item["config"],
                "metrics": metrics,
                "is_new": item.get("is_new", False),
            }
        )

    report = build_report(results, df)
    REPORT_FILE.write_text(report, encoding="utf-8")

    for item in results:
        metrics = item["metrics"]
        print(
            f"{item['name']}: final={metrics['final_value']:.2f}, "
            f"total_return={metrics['total_return_pct']:.2f}%, "
            f"annual_return={metrics['annual_return_pct']:.2f}%, "
            f"sharpe={metric_text(metrics['sharpe'])}, "
            f"max_dd={metrics['max_drawdown_pct']:.2f}%, "
            f"closed_trades={metrics['closed_trades']}, "
            f"win_rate={metrics['win_rate_pct']:.2f}%, "
            f"avg_exposure={metrics['exposure_pct']:.2f}%"
        )
    print(f"Report written to {REPORT_FILE.name}")
    return results


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--optimize-dualthrust",
        action="store_true",
        help="Run a focused parameter search for DualThrustRegimeStrategy and write a markdown summary.",
    )
    return parser.parse_args()


def main():
    args = parse_args()
    if args.optimize_dualthrust:
        run_dualthrust_optimization()
        return
    run_experiments()


if __name__ == "__main__":
    main()