#!/usr/bin/env python3 """补录缺失的趋势回调策略结束快照（strategy_trade_snapshots）。 适用：gate_bot 等在计划结束（止盈/止损/手动）时因 strategy_trend_cfg 未注册而漏写快照的历史数据。 保本移交路径通常已有快照，本脚本默认跳过「已有任意快照」的计划。 用法（在仓库根目录，Linux 请用 python3）： python3 scripts/backfill_trend_strategy_snapshots.py \\ --db crypto_monitor_gate_bot/crypto.db --dry-run python3 scripts/backfill_trend_strategy_snapshots.py \\ --db crypto_monitor_gate_bot/crypto.db --apply """ from __future__ import annotations import argparse import sqlite3 import sys from pathlib import Path _REPO_ROOT = Path(__file__).resolve().parents[1] if str(_REPO_ROOT) not in sys.path: sys.path.insert(0, str(_REPO_ROOT)) from strategy_snapshot_lib import ( # noqa: E402 STRATEGY_TREND, init_strategy_snapshot_table, save_trend_plan_snapshot, ) PLAN_STATUS_LABEL = { "stopped_sl": "止损", "stopped_tp": "止盈", "stopped_manual": "手动平仓", "stopped_handoff": "保本移交", } TRADE_RESULT_LABEL = { "止损": "止损", "止盈": "止盈", "手动平仓": "手动平仓", "移动止盈": "止盈", "保本止盈": "止盈", "强制清仓": "手动平仓", } def _row_dict(row) -> dict: if row is None: return {} try: return dict(row) except Exception: return {} def infer_exit_price( direction: str, entry: float | None, margin: float | None, leverage: float | None, pnl: float | None, ) -> float | None: """由本地 calc_pnl 口径反推平仓价（供补录快照 exit_price）。""" try: trigger = float(entry) margin_f = float(margin) lev = float(leverage) pnl_f = float(pnl) except (TypeError, ValueError): return None if trigger <= 0 or margin_f <= 0 or lev <= 0: return None notional = margin_f * lev if notional <= 0: return None ratio = pnl_f / notional if (direction or "long").strip().lower() == "short": return round(trigger * (1.0 - ratio), 10) return round(trigger * (1.0 + ratio), 10) def resolve_result_label(plan: dict, trade: dict | None) -> str: status = (plan.get("status") or "").strip() if status in PLAN_STATUS_LABEL: return PLAN_STATUS_LABEL[status] if trade: res = (trade.get("result") or "").strip() if res in TRADE_RESULT_LABEL: return TRADE_RESULT_LABEL[res] if res: return res msg = (plan.get("message") or "").strip() if msg: return msg[:32] return "结束" def find_missing_plans( conn: sqlite3.Connection, *, plan_id: int | None = None, since: str | None = None, ) -> list[dict]: sql = """ SELECT p.* FROM trend_pullback_plans p WHERE TRIM(COALESCE(p.status, '')) != 'active' AND NOT EXISTS ( SELECT 1 FROM strategy_trade_snapshots s WHERE s.strategy_type = ? AND s.source_id = p.id ) """ params: list[object] = [STRATEGY_TREND] if plan_id is not None: sql += " AND p.id = ?" params.append(int(plan_id)) if since: sql += " AND COALESCE(p.opened_at, '') >= ?" params.append(since.strip()) sql += " ORDER BY p.id ASC" rows = conn.execute(sql, params).fetchall() return [_row_dict(r) for r in rows] def fetch_trade_for_plan(conn: sqlite3.Connection, plan_id: int) -> dict | None: row = conn.execute( """ SELECT * FROM trade_records WHERE trend_plan_id = ? ORDER BY COALESCE(closed_at_ms, 0) DESC, id DESC LIMIT 1 """, (int(plan_id),), ).fetchone() return _row_dict(row) if row else None def backfill_one(conn: sqlite3.Connection, plan: dict, *, dry_run: bool) -> dict: plan_id = int(plan["id"]) trade = fetch_trade_for_plan(conn, plan_id) result_label = resolve_result_label(plan, trade) pnl_amount = None closed_at = None exit_price = None entry = plan.get("avg_entry_price") or plan.get("live_price_ref") margin = plan.get("plan_margin_capital") leverage = plan.get("leverage") if trade: pnl_amount = trade.get("pnl_amount") closed_at = trade.get("closed_at") entry = trade.get("trigger_price") or entry margin = trade.get("margin_capital") or margin leverage = trade.get("leverage") or leverage exit_price = infer_exit_price( plan.get("direction") or trade.get("direction") or "long", entry, margin, leverage, pnl_amount, ) info = { "plan_id": plan_id, "symbol": plan.get("symbol"), "status": plan.get("status"), "result_label": result_label, "closed_at": closed_at, "pnl_amount": pnl_amount, "exit_price": exit_price, "legs_done": plan.get("legs_done"), "dca_legs": plan.get("dca_legs"), "has_trade": bool(trade), } if dry_run: return info save_trend_plan_snapshot( {}, conn, plan, result_label=result_label, exit_price=exit_price, pnl_amount=float(pnl_amount) if pnl_amount is not None else None, closed_at=closed_at, ) return info def main() -> int: parser = argparse.ArgumentParser( description="Backfill missing trend_pullback strategy_trade_snapshots rows." ) parser.add_argument("--db", required=True, help="Path to instance sqlite db") parser.add_argument("--plan-id", type=int, help="Only backfill this trend plan id") parser.add_argument( "--since", help="Only plans with opened_at >= YYYY-MM-DD (optional)", ) parser.add_argument("--dry-run", action="store_true", help="Preview only (default)") parser.add_argument("--apply", action="store_true", help="Write snapshots") args = parser.parse_args() if not args.dry_run and not args.apply: args.dry_run = True db_path = Path(args.db).expanduser().resolve() if not db_path.is_file(): print(f"[ERR] DB not found: {db_path}") return 1 conn = sqlite3.connect(str(db_path)) conn.row_factory = sqlite3.Row init_strategy_snapshot_table(conn) missing = find_missing_plans( conn, plan_id=args.plan_id, since=args.since ) if not missing: print("[INFO] No closed trend plans missing strategy snapshots.") conn.close() return 0 print(f"[INFO] Found {len(missing)} plan(s) without strategy snapshot.") applied = 0 for plan in missing: info = backfill_one(conn, plan, dry_run=not args.apply) trade_hint = "有交易记录" if info["has_trade"] else "无交易记录" print( f" - plan #{info['plan_id']} {info['symbol']} " f"status={info['status']} → {info['result_label']} " f"closed={info['closed_at'] or '—'} pnl={info['pnl_amount']} " f"补仓 {info['legs_done']}/{info['dca_legs']} ({trade_hint})" ) applied += 1 if args.apply: conn.commit() print(f"[OK] Backfilled {applied} snapshot(s).") else: print("[DRY-RUN] No changes written. Re-run with --apply to commit.") conn.close() return 0 if __name__ == "__main__": raise SystemExit(main())