cyb50-quant/research/dragon/v2/dragon_threshold_perturbation.py

from __future__ import annotations

from pathlib import Path

import pandas as pd

from dragon_indicators import DragonIndicatorConfig, DragonIndicatorEngine
from dragon_strategy import DragonRuleEngine
from dragon_strategy_config import StrategyConfig
from dragon_workbook import DragonWorkbook


def _find_workbook(base_dir: Path) -> Path:
    matches = sorted(base_dir.glob("*.xlsx"))
    if not matches:
        raise FileNotFoundError(f"No workbook found in {base_dir}")
    return matches[0]


def _load_workbook_events(workbook_path: Path) -> pd.DataFrame:
    workbook = DragonWorkbook(workbook_path)
    return pd.DataFrame(
        [
            {
                "date": event.date.isoformat(),
                "side": event.side,
                "layer": event.layer,
            }
            for event in workbook.split_layers()
        ]
    )


def _event_overlap(workbook_events: pd.DataFrame, strategy_events: pd.DataFrame, side: str, layer: str) -> tuple[int, int, int]:
    wb = set(workbook_events[(workbook_events["side"] == side) & (workbook_events["layer"] == layer)]["date"])
    st = set(strategy_events[(strategy_events["side"] == side) & (strategy_events["layer"] == layer)]["date"])
    hit = wb & st
    return len(hit), len(wb - st), len(st - wb)


def _profit_factor(series: pd.Series) -> float:
    gross_profit = series[series > 0].sum()
    gross_loss = -series[series < 0].sum()
    if gross_loss == 0:
        return float("inf") if gross_profit > 0 else 0.0
    return float(gross_profit / gross_loss)


def _run_experiment(
    label: str,
    parameter: str,
    variant: str,
    config: StrategyConfig,
    workbook_events: pd.DataFrame,
    indicator_df: pd.DataFrame,
    first_workbook_date: str,
    last_workbook_date: str,
) -> dict[str, object]:
    strategy = DragonRuleEngine(config=config)
    events, trades = strategy.run(indicator_df)
    events = events[(events["date"] >= first_workbook_date) & (events["date"] <= last_workbook_date)].copy()
    trades = trades[
        (trades["buy_date"] >= first_workbook_date)
        & (trades["buy_date"] <= last_workbook_date)
        & (trades["sell_date"] >= first_workbook_date)
        & (trades["sell_date"] <= last_workbook_date)
    ].copy()

    real_buy_overlap, real_buy_missing, real_buy_extra = _event_overlap(workbook_events, events, "BUY", "real_trade")
    real_sell_overlap, real_sell_missing, real_sell_extra = _event_overlap(workbook_events, events, "SELL", "real_trade")
    aux_sell_overlap, aux_sell_missing, aux_sell_extra = _event_overlap(workbook_events, events, "SELL", "aux_signal")

    return {
        "experiment": label,
        "parameter": parameter,
        "variant": variant,
        "value": getattr(config, parameter) if parameter != "baseline" else "baseline",
        "trades": int(len(trades)),
        "win_rate": float((trades["return_pct"] > 0).mean()) if not trades.empty else float("nan"),
        "avg_return": float(trades["return_pct"].mean()) if not trades.empty else float("nan"),
        "profit_factor": _profit_factor(trades["return_pct"]) if not trades.empty else float("nan"),
        "real_buy_overlap": int(real_buy_overlap),
        "real_buy_missing": int(real_buy_missing),
        "real_buy_extra": int(real_buy_extra),
        "real_sell_overlap": int(real_sell_overlap),
        "real_sell_missing": int(real_sell_missing),
        "real_sell_extra": int(real_sell_extra),
        "aux_sell_overlap": int(aux_sell_overlap),
        "aux_sell_missing": int(aux_sell_missing),
        "aux_sell_extra": int(aux_sell_extra),
    }


def main() -> None:
    base_dir = Path(__file__).resolve().parent
    workbook_path = _find_workbook(base_dir)
    workbook_events = _load_workbook_events(workbook_path)
    first_workbook_date = pd.to_datetime(workbook_events["date"]).min().date().isoformat()
    last_workbook_date = pd.to_datetime(workbook_events["date"]).max().date().isoformat()

    engine = DragonIndicatorEngine(DragonIndicatorConfig(start_date="2015-01-01", end_date="2026-01-31"))
    indicator_df = engine.compute(engine.fetch_daily_data())

    baseline = StrategyConfig()
    perturbations = {
        "post_exit_confirmation_window_days": [8, 10, 12],
        "aux_sell_high_zone_kdj_only_block_c1": [83.0, 85.0, 87.0],
        "glued_high_weak_rebound_high_c1": [66.0, 68.0, 70.0],
        "glued_high_weak_rebound_high_b1": [-0.10, -0.08, -0.06],
        "deep_oversold_entry_c1_max": [15.0, 16.0, 17.0],
        "deep_oversold_entry_b1_min": [-0.12, -0.10, -0.08],
        "predictive_b1_break_short_b1_max": [-0.15, -0.13, -0.11],
        "predictive_b1_break_long_b1_max": [-0.14, -0.12, -0.10],
    }

    rows = [
        _run_experiment(
            "baseline",
            "baseline",
            "baseline",
            baseline,
            workbook_events,
            indicator_df,
            first_workbook_date,
            last_workbook_date,
        )
    ]

    for parameter, values in perturbations.items():
        for idx, value in enumerate(values):
            variant = "lower" if idx == 0 else "baseline" if idx == 1 else "upper"
            config = baseline.with_updates(**{parameter: value})
            rows.append(
                _run_experiment(
                    f"{parameter}:{variant}",
                    parameter,
                    variant,
                    config,
                    workbook_events,
                    indicator_df,
                    first_workbook_date,
                    last_workbook_date,
                )
            )

    result_df = pd.DataFrame(rows)
    baseline_row = result_df[result_df["experiment"] == "baseline"].iloc[0]
    for col in ["trades", "win_rate", "avg_return", "profit_factor", "real_buy_overlap", "real_sell_overlap", "aux_sell_overlap"]:
        result_df[f"delta_{col}"] = result_df[col] - baseline_row[col]

    result_df.to_csv(base_dir / "dragon_threshold_perturbation.csv", index=False, encoding="utf-8-sig")

    summary_rows: list[dict[str, object]] = []
    for parameter in perturbations:
        subset = result_df[result_df["parameter"] == parameter].copy()
        summary_rows.append(
            {
                "parameter": parameter,
                "baseline_value": subset[subset["variant"] == "baseline"].iloc[0]["value"],
                "real_buy_overlap_min": int(subset["real_buy_overlap"].min()),
                "real_sell_overlap_min": int(subset["real_sell_overlap"].min()),
                "avg_return_range": float(subset["avg_return"].max() - subset["avg_return"].min()),
                "profit_factor_range": float(subset["profit_factor"].max() - subset["profit_factor"].min()),
                "aux_sell_overlap_range": int(subset["aux_sell_overlap"].max() - subset["aux_sell_overlap"].min()),
                "stable_real_alignment": bool(
                    (subset["real_buy_overlap"] == baseline_row["real_buy_overlap"]).all()
                    and (subset["real_sell_overlap"] == baseline_row["real_sell_overlap"]).all()
                ),
            }
        )
    summary_df = pd.DataFrame(summary_rows).sort_values(["stable_real_alignment", "avg_return_range"], ascending=[True, False])
    summary_df.to_csv(base_dir / "dragon_threshold_sensitivity_summary.csv", index=False, encoding="utf-8-sig")

    lines = [
        "# Dragon Threshold Perturbation",
        "",
        "## Baseline",
        f"- trades: `{int(baseline_row['trades'])}`",
        f"- avg_return: `{baseline_row['avg_return']:.2%}`",
        f"- profit_factor: `{baseline_row['profit_factor']:.2f}`",
        f"- real BUY overlap: `{int(baseline_row['real_buy_overlap'])}`",
        f"- real SELL overlap: `{int(baseline_row['real_sell_overlap'])}`",
        "",
        "## Sensitivity Summary",
    ]
    for _, row in summary_df.iterrows():
        lines.append(
            f"- `{row['parameter']}`: stable_real_alignment `{bool(row['stable_real_alignment'])}`, "
            f"avg_return_range `{row['avg_return_range']:.2%}`, profit_factor_range `{row['profit_factor_range']:.2f}`, "
            f"aux_sell_overlap_range `{int(row['aux_sell_overlap_range'])}`"
        )

    fragile = summary_df[~summary_df["stable_real_alignment"]]
    lines.extend(["", "## Fragile Parameters"])
    if fragile.empty:
        lines.append("- None in this first perturbation pack.")
    else:
        for _, row in fragile.iterrows():
            lines.append(
                f"- `{row['parameter']}`: minimum real BUY overlap `{int(row['real_buy_overlap_min'])}`, minimum real SELL overlap `{int(row['real_sell_overlap_min'])}`"
            )

    robust = summary_df[summary_df["stable_real_alignment"]].sort_values("avg_return_range").head(6)
    lines.extend(["", "## Relatively Robust Parameters"])
    for _, row in robust.iterrows():
        lines.append(
            f"- `{row['parameter']}`: avg_return_range `{row['avg_return_range']:.2%}`, profit_factor_range `{row['profit_factor_range']:.2f}`"
        )

    (base_dir / "dragon_threshold_perturbation.md").write_text("\n".join(lines) + "\n", encoding="utf-8")


if __name__ == "__main__":
    main()