from __future__ import annotations from pathlib import Path from typing import Optional import pandas as pd from dragon_branch_configs import alpha_first_glued_selective_veto_config, alpha_first_selective_veto_config from dragon_strategy import DragonRuleEngine def _load_csv(base_dir: Path, name: str) -> pd.DataFrame: return pd.read_csv(base_dir / name, encoding="utf-8-sig") def _load_indicator_snapshot(base_dir: Path) -> pd.DataFrame: df = _load_csv(base_dir, "dragon_indicator_snapshot.csv") df["date"] = pd.to_datetime(df["date"]) return df.sort_values("date").reset_index(drop=True) 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 _pct(value: Optional[float]) -> str: if value is None or pd.isna(value): return "n/a" if value == float("inf"): return "inf" return f"{float(value):.2%}" def _fmt_num(value: Optional[float]) -> str: if value is None or pd.isna(value): return "n/a" if value == float("inf"): return "inf" return f"{float(value):.2f}" def _holding_bucket(days: int) -> str: if days <= 5: return "00-05d" if days <= 10: return "06-10d" if days <= 20: return "11-20d" if days <= 40: return "21-40d" return "41d+" def _build_trade_quality(trades: pd.DataFrame, indicators: pd.DataFrame) -> pd.DataFrame: trades = trades.copy() trades["buy_dt"] = pd.to_datetime(trades["buy_date"]) trades["sell_dt"] = pd.to_datetime(trades["sell_date"]) trades["holding_bucket"] = trades["holding_days"].astype(int).map(_holding_bucket) pos_lookup = {dt.date().isoformat(): idx for idx, dt in enumerate(indicators["date"])} buy_a1: list[float] = [] buy_b1: list[float] = [] buy_c1: list[float] = [] sell_a1: list[float] = [] sell_b1: list[float] = [] sell_c1: list[float] = [] mfe_list: list[float] = [] mae_list: list[float] = [] exit_followthrough_list: list[float] = [] exit_rebound_list: list[float] = [] entry_forward_5d_list: list[float] = [] indicator_by_date = indicators.set_index(indicators["date"].dt.date) for _, trade in trades.iterrows(): buy_date = pd.Timestamp(trade["buy_date"]).date() sell_date = pd.Timestamp(trade["sell_date"]).date() entry_price = float(trade["buy_price"]) exit_price = float(trade["sell_price"]) buy_row = indicator_by_date.loc[buy_date] sell_row = indicator_by_date.loc[sell_date] buy_a1.append(float(buy_row["a1"])) buy_b1.append(float(buy_row["b1"])) buy_c1.append(float(buy_row["c1"])) sell_a1.append(float(sell_row["a1"])) sell_b1.append(float(sell_row["b1"])) sell_c1.append(float(sell_row["c1"])) window = indicators[(indicators["date"] >= pd.Timestamp(trade["buy_date"])) & (indicators["date"] <= pd.Timestamp(trade["sell_date"]))] max_high = float(window["high"].max()) min_low = float(window["low"].min()) mfe_list.append(max_high / entry_price - 1.0) mae_list.append(min_low / entry_price - 1.0) sell_idx = pos_lookup[trade["sell_date"]] future = indicators.iloc[sell_idx + 1 : sell_idx + 6] if future.empty: exit_followthrough_list.append(float("nan")) exit_rebound_list.append(float("nan")) else: exit_followthrough_list.append(float(future["low"].min()) / exit_price - 1.0) exit_rebound_list.append(float(future["high"].max()) / exit_price - 1.0) buy_idx = pos_lookup[trade["buy_date"]] entry_future = indicators.iloc[buy_idx + 1 : buy_idx + 6] if entry_future.empty: entry_forward_5d_list.append(float("nan")) else: entry_forward_5d_list.append(float(entry_future["close"].iloc[-1]) / entry_price - 1.0) trades["buy_a1"] = buy_a1 trades["buy_b1"] = buy_b1 trades["buy_c1"] = buy_c1 trades["sell_a1"] = sell_a1 trades["sell_b1"] = sell_b1 trades["sell_c1"] = sell_c1 trades["mfe_pct"] = mfe_list trades["mae_pct"] = mae_list trades["exit_followthrough_5d_pct"] = exit_followthrough_list trades["exit_rebound_5d_pct"] = exit_rebound_list trades["entry_forward_5d_pct"] = entry_forward_5d_list return trades def _run_branch(indicators: pd.DataFrame, config) -> pd.DataFrame: indicator_indexed = indicators.set_index("date", drop=False) engine = DragonRuleEngine(config=config) _, trades = engine.run(indicator_indexed) trades = trades.copy() return _build_trade_quality(trades, indicators) def _trade_key(df: pd.DataFrame) -> set[tuple[str, str, str, str]]: return set(zip(df["buy_date"], df["sell_date"], df["buy_reason"], df["sell_reason"])) def _veto_bucket(c1: float, b1: float) -> str: if c1 >= 40 and b1 >= 0.10: return "hot_positive_b1" if 23 <= c1 < 28 and b1 <= 0.02: return "low_weak_range" return "other" def _recommendation(row: pd.Series) -> tuple[str, str]: ret = float(row["return_pct"]) holding = int(row["holding_days"]) mfe = float(row["mfe_pct"]) replacement_ret = row.get("replacement_return_pct") if ret > 0: if ret <= 0.01 and holding <= 15: if pd.notna(replacement_ret) and float(replacement_ret) <= ret: return "OBSERVE_VETO", "原交易仅微利，但当前替代路径没有更强，建议进一步细化 hot 过滤而不是直接全盘保留。" return "OBSERVE_VETO", "原交易为微利短单，暂不应视为强 alpha，先保留为观察样本。" return "OVER_VETO", "原交易本身为明显盈利单，当前过滤过度。" if holding <= 10 and ret < 0 and float(row["exit_followthrough_5d_pct"]) <= 0: return "KEEP_VETO", "短持仓亏损且卖出后继续走弱，属于应优先清理的噪音单。" if ret < 0 and mfe <= 0.02: return "KEEP_VETO", "持仓期间几乎没有有效盈利空间，删除逻辑合理。" return "KEEP_VETO", "总体为负收益短单，保留 veto 更符合当前 alpha-first 目标。" def _summary_line(group: pd.DataFrame) -> str: if group.empty: return "n/a" return ( f"trades `{len(group)}`, win_rate `{_pct(float((group['return_pct'] > 0).mean()))}`, " f"avg_return `{_pct(float(group['return_pct'].mean()))}`, avg_mfe `{_pct(float(group['mfe_pct'].mean()))}`, " f"avg_mae `{_pct(float(group['mae_pct'].mean()))}`, avg_holding `{float(group['holding_days'].mean()):.2f}`" ) def main() -> None: base_dir = Path(__file__).resolve().parent indicators = _load_indicator_snapshot(base_dir) workbook_events = _load_csv(base_dir, "true_trade_events.csv") alpha_trades = _run_branch(indicators, alpha_first_selective_veto_config()) glued_trades = _run_branch(indicators, alpha_first_glued_selective_veto_config()) workbook_buy = set(workbook_events[(workbook_events["layer"] == "real_trade") & (workbook_events["side"] == "BUY")]["date"]) workbook_sell = set(workbook_events[(workbook_events["layer"] == "real_trade") & (workbook_events["side"] == "SELL")]["date"]) alpha_set = _trade_key(alpha_trades) glued_set = _trade_key(glued_trades) removed = pd.DataFrame( sorted(alpha_set - glued_set), columns=["buy_date", "sell_date", "buy_reason", "sell_reason"], ) rows: list[dict[str, object]] = [] for _, removed_row in removed.iterrows(): match = alpha_trades[ (alpha_trades["buy_date"] == removed_row["buy_date"]) & (alpha_trades["sell_date"] == removed_row["sell_date"]) & (alpha_trades["buy_reason"] == removed_row["buy_reason"]) & (alpha_trades["sell_reason"] == removed_row["sell_reason"]) ] if match.empty: continue trade = match.iloc[0] sell_dt = pd.Timestamp(trade["sell_date"]) replacement = glued_trades[ (pd.to_datetime(glued_trades["buy_date"]) > sell_dt) & (pd.to_datetime(glued_trades["buy_date"]) <= sell_dt + pd.Timedelta(days=10)) ].sort_values("buy_date") replacement_row = replacement.iloc[0] if not replacement.empty else None rec, rec_reason = _recommendation( pd.Series( { **trade.to_dict(), "replacement_return_pct": None if replacement_row is None else float(replacement_row["return_pct"]), } ) ) rows.append( { "buy_date": trade["buy_date"], "sell_date": trade["sell_date"], "buy_reason": trade["buy_reason"], "sell_reason": trade["sell_reason"], "holding_bucket": trade["holding_bucket"], "holding_days": int(trade["holding_days"]), "return_pct": float(trade["return_pct"]), "mfe_pct": float(trade["mfe_pct"]), "mae_pct": float(trade["mae_pct"]), "entry_forward_5d_pct": float(trade["entry_forward_5d_pct"]), "exit_followthrough_5d_pct": float(trade["exit_followthrough_5d_pct"]), "exit_rebound_5d_pct": float(trade["exit_rebound_5d_pct"]), "buy_a1": float(trade["buy_a1"]), "buy_b1": float(trade["buy_b1"]), "buy_c1": float(trade["buy_c1"]), "sell_a1": float(trade["sell_a1"]), "sell_b1": float(trade["sell_b1"]), "sell_c1": float(trade["sell_c1"]), "veto_bucket": _veto_bucket(float(trade["buy_c1"]), float(trade["buy_b1"])), "buy_aligned_with_workbook": trade["buy_date"] in workbook_buy, "sell_aligned_with_workbook": trade["sell_date"] in workbook_sell, "replacement_buy_date": "" if replacement_row is None else str(replacement_row["buy_date"]), "replacement_sell_date": "" if replacement_row is None else str(replacement_row["sell_date"]), "replacement_buy_reason": "" if replacement_row is None else str(replacement_row["buy_reason"]), "replacement_sell_reason": "" if replacement_row is None else str(replacement_row["sell_reason"]), "replacement_return_pct": float("nan") if replacement_row is None else float(replacement_row["return_pct"]), "replacement_gap_days": float("nan") if replacement_row is None else int((pd.Timestamp(replacement_row["buy_date"]) - sell_dt).days), "recommendation": rec, "recommendation_reason": rec_reason, } ) attribution = pd.DataFrame(rows).sort_values(["veto_bucket", "buy_date"]).reset_index(drop=True) attribution.to_csv(base_dir / "dragon_glued_veto_attribution.csv", index=False, encoding="utf-8-sig") bucket_summary = ( attribution.groupby("veto_bucket", dropna=False) .agg( trades=("buy_date", "count"), win_rate=("return_pct", lambda s: float((s > 0).mean())), avg_return=("return_pct", "mean"), avg_mfe=("mfe_pct", "mean"), avg_mae=("mae_pct", "mean"), avg_holding_days=("holding_days", "mean"), keep_veto_count=("recommendation", lambda s: int((s == "KEEP_VETO").sum())), observe_veto_count=("recommendation", lambda s: int((s == "OBSERVE_VETO").sum())), over_veto_count=("recommendation", lambda s: int((s == "OVER_VETO").sum())), ) .reset_index() .sort_values(["trades", "avg_return"], ascending=[False, True]) ) bucket_summary.to_csv(base_dir / "dragon_glued_veto_bucket_summary.csv", index=False, encoding="utf-8-sig") lines = [ "# Dragon Glued Veto Attribution Review", "", "## Snapshot", f"- removed trades vs current alpha-first: `{len(attribution)}`", f"- total avg_return of removed set: `{_pct(float(attribution['return_pct'].mean()))}`", f"- total win_rate of removed set: `{_pct(float((attribution['return_pct'] > 0).mean()))}`", f"- removed-set profit_factor: `{_fmt_num(_profit_factor(attribution['return_pct']))}`", "", "## Bucket Summary", ] for _, row in bucket_summary.iterrows(): lines.append( f"- `{row['veto_bucket']}`: trades `{int(row['trades'])}`, win_rate `{_pct(float(row['win_rate']))}`, " f"avg_return `{_pct(float(row['avg_return']))}`, avg_mfe `{_pct(float(row['avg_mfe']))}`, " f"avg_mae `{_pct(float(row['avg_mae']))}`, avg_holding `{float(row['avg_holding_days']):.2f}`, " f"KEEP/OBSERVE/OVER = `{int(row['keep_veto_count'])}/{int(row['observe_veto_count'])}/{int(row['over_veto_count'])}`" ) lines.extend( [ "", "## Quant Judgment", f"- `low_weak_range`: {_summary_line(attribution[attribution['veto_bucket'] == 'low_weak_range'])}.", f"- `hot_positive_b1`: {_summary_line(attribution[attribution['veto_bucket'] == 'hot_positive_b1'])}.", "- `low_weak_range` is now a clean promotion candidate: all removed trades are short, losing, and there is no positive sample in this bucket.", "- `hot_positive_b1` is directionally correct but not fully clean: most removed trades are weak, but one micro-profit sample remains and should be used as the first refinement target.", "- Immediate next research action: keep the low bucket intact, and narrow the hot bucket rather than rolling back the whole glued veto branch.", "", "## Detailed Cards", ] ) for _, row in attribution.iterrows(): replacement = "none" if isinstance(row["replacement_buy_date"], str) and row["replacement_buy_date"]: replacement = ( f"{row['replacement_buy_date']} -> {row['replacement_sell_date']} / " f"{row['replacement_buy_reason']} -> {row['replacement_sell_reason']} / " f"{_pct(row['replacement_return_pct'])}" ) lines.extend( [ f"### {row['buy_date']} -> {row['sell_date']}", f"- Bucket: `{row['veto_bucket']}`", f"- Trade: `{row['buy_reason']} -> {row['sell_reason']}` | `{int(row['holding_days'])}` days | return `{_pct(row['return_pct'])}`", f"- MFE / MAE: `{_pct(row['mfe_pct'])}` / `{_pct(row['mae_pct'])}`", f"- Entry 5d / Exit followthrough 5d: `{_pct(row['entry_forward_5d_pct'])}` / `{_pct(row['exit_followthrough_5d_pct'])}`", f"- Entry indicators: `a1={float(row['buy_a1']):.4f}` `b1={float(row['buy_b1']):.4f}` `c1={float(row['buy_c1']):.2f}`", f"- Workbook aligned: buy `{bool(row['buy_aligned_with_workbook'])}` / sell `{bool(row['sell_aligned_with_workbook'])}`", f"- Candidate replacement within 10d after exit: `{replacement}`", f"- Recommendation: `{row['recommendation']}` | {row['recommendation_reason']}", "", ] ) (base_dir / "dragon_glued_veto_review.md").write_text("\n".join(lines) + "\n", encoding="utf-8") if __name__ == "__main__": main()