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()