cyb50-quant/research/dragon/v2/dragon_predictive_break_audit.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


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 main() -> None:
    base_dir = Path(__file__).resolve().parent
    engine = DragonIndicatorEngine(DragonIndicatorConfig(start_date="2015-01-01", end_date="2026-01-31"))
    indicator_df = engine.compute(engine.fetch_daily_data())

    baseline_events, baseline_trades = DragonRuleEngine(config=StrategyConfig()).run(indicator_df)
    disabled_events, disabled_trades = DragonRuleEngine(
        config=StrategyConfig(disabled_rules=frozenset({"predictive_b1_break_exit"}))
    ).run(indicator_df)

    predictive = baseline_trades[baseline_trades["sell_reason"] == "predictive_b1_break_exit"].copy()
    if predictive.empty:
        raise RuntimeError("No predictive_b1_break_exit trades found in baseline.")

    predictive["buy_dt"] = pd.to_datetime(predictive["buy_date"])
    predictive["sell_dt"] = pd.to_datetime(predictive["sell_date"])
    indicator_df["dt"] = indicator_df.index
    indicator_df = indicator_df.reset_index(drop=True)
    pos_lookup = {dt.date().isoformat(): idx for idx, dt in enumerate(pd.to_datetime(indicator_df["dt"]))}

    rows: list[dict[str, object]] = []
    for _, trade in predictive.iterrows():
        sell_idx = pos_lookup[trade["sell_date"]]
        entry_price = float(trade["buy_price"])
        exit_price = float(trade["sell_price"])

        future_3 = indicator_df.iloc[sell_idx + 1 : sell_idx + 4]
        future_5 = indicator_df.iloc[sell_idx + 1 : sell_idx + 6]
        future_10 = indicator_df.iloc[sell_idx + 1 : sell_idx + 11]

        alt_trade = disabled_trades[
            (disabled_trades["buy_date"] == trade["buy_date"]) & (disabled_trades["buy_reason"] == trade["buy_reason"])
        ].copy()
        alt_sell_date = alt_trade["sell_date"].iloc[0] if not alt_trade.empty else ""
        alt_sell_reason = alt_trade["sell_reason"].iloc[0] if not alt_trade.empty else ""
        alt_return = float(alt_trade["return_pct"].iloc[0]) if not alt_trade.empty else float("nan")

        resumed_trade = baseline_trades[baseline_trades["buy_date"] > trade["sell_date"]].head(1).copy()
        resumed_buy_date = resumed_trade["buy_date"].iloc[0] if not resumed_trade.empty else ""
        resumed_buy_reason = resumed_trade["buy_reason"].iloc[0] if not resumed_trade.empty else ""
        resumed_sell_date = resumed_trade["sell_date"].iloc[0] if not resumed_trade.empty else ""
        resumed_return = float(resumed_trade["return_pct"].iloc[0]) if not resumed_trade.empty else float("nan")

        combined_return = (1.0 + float(trade["return_pct"])) * (1.0 + resumed_return) - 1.0 if resumed_trade.shape[0] == 1 else float("nan")

        rows.append(
            {
                "buy_date": trade["buy_date"],
                "buy_reason": trade["buy_reason"],
                "predictive_sell_date": trade["sell_date"],
                "predictive_return_pct": float(trade["return_pct"]),
                "predictive_holding_days": int(trade["holding_days"]),
                "exit_a1": float(
                    baseline_events[
                        (baseline_events["date"] == trade["sell_date"])
                        & (baseline_events["layer"] == "real_trade")
                        & (baseline_events["side"] == "SELL")
                    ]["a1"].iloc[0]
                ),
                "exit_b1": float(
                    baseline_events[
                        (baseline_events["date"] == trade["sell_date"])
                        & (baseline_events["layer"] == "real_trade")
                        & (baseline_events["side"] == "SELL")
                    ]["b1"].iloc[0]
                ),
                "exit_c1": float(
                    baseline_events[
                        (baseline_events["date"] == trade["sell_date"])
                        & (baseline_events["layer"] == "real_trade")
                        & (baseline_events["side"] == "SELL")
                    ]["c1"].iloc[0]
                ),
                "future_3d_low_pct": float(future_3["low"].min()) / exit_price - 1.0 if not future_3.empty else float("nan"),
                "future_3d_high_pct": float(future_3["high"].max()) / exit_price - 1.0 if not future_3.empty else float("nan"),
                "future_5d_low_pct": float(future_5["low"].min()) / exit_price - 1.0 if not future_5.empty else float("nan"),
                "future_5d_high_pct": float(future_5["high"].max()) / exit_price - 1.0 if not future_5.empty else float("nan"),
                "future_10d_low_pct": float(future_10["low"].min()) / exit_price - 1.0 if not future_10.empty else float("nan"),
                "future_10d_high_pct": float(future_10["high"].max()) / exit_price - 1.0 if not future_10.empty else float("nan"),
                "disabled_alt_sell_date": alt_sell_date,
                "disabled_alt_sell_reason": alt_sell_reason,
                "disabled_alt_return_pct": alt_return,
                "resumed_buy_date": resumed_buy_date,
                "resumed_buy_reason": resumed_buy_reason,
                "resumed_sell_date": resumed_sell_date,
                "resumed_return_pct": resumed_return,
                "combined_split_return_pct": combined_return,
            }
        )

    audit_df = pd.DataFrame(rows)
    audit_df.to_csv(base_dir / "dragon_predictive_break_audit.csv", index=False, encoding="utf-8-sig")

    lines = [
        "# Dragon Predictive Break Review",
        "",
        f"- predictive exit trades in baseline: `{len(audit_df)}`",
        f"- predictive exit avg return: `{audit_df['predictive_return_pct'].mean():.2%}`",
        f"- disabled-rule alt avg return: `{audit_df['disabled_alt_return_pct'].mean():.2%}`",
        f"- split-chain combined avg return: `{audit_df['combined_split_return_pct'].mean():.2%}`",
        "",
        "## Trade Cards",
    ]
    for _, row in audit_df.iterrows():
        lines.append(
            f"- `{row['buy_date']} -> {row['predictive_sell_date']}` `{row['buy_reason']}` | "
            f"exit a1 `{row['exit_a1']:.4f}` b1 `{row['exit_b1']:.4f}` c1 `{row['exit_c1']:.2f}` | "
            f"3d low `{row['future_3d_low_pct']:.2%}` / 5d high `{row['future_5d_high_pct']:.2%}`"
        )
        lines.append(
            f"  disabled path -> `{row['disabled_alt_sell_date']}` `{row['disabled_alt_sell_reason']}` `{row['disabled_alt_return_pct']:.2%}`"
        )
        lines.append(
            f"  split path reentry -> `{row['resumed_buy_date']}` `{row['resumed_buy_reason']}` to `{row['resumed_sell_date']}` `{row['resumed_return_pct']:.2%}`; combined `{row['combined_split_return_pct']:.2%}`"
        )

    lines.extend(
        [
            "",
            "## Quant Judgment",
            "- The remaining predictive break is a bridge-style exit, not a generic stop-loss bucket.",
            "- Disabling it improves the single uninterrupted trade return, but destroys the workbook-aligned split chain.",
            "- Under the current reconstruction objective, this rule should be frozen unless the user explicitly accepts lower workbook alignment.",
        ]
    )
    (base_dir / "dragon_predictive_break_review.md").write_text("\n".join(lines) + "\n", encoding="utf-8")


if __name__ == "__main__":
    main()