# -*- coding: utf-8 -*- """ 策略判分读数 · 只读统计 (STRATEGY_AUTO_ATTACH_PLAN.md 判分闭环, 拍板⑤) ====================================================================== 只读, 不写任何表, 随时可跑。回答一个问题: **自动挂载的策略到底有没有挣到钱**。 三张对照 + 一张盘点: 零. 样本盘点 自动策略共几条 (在跑/撤下/接力), 台账留痕几条 —— 先看样本够不够 一. 网格差价 每条自动网格: 买了多少/卖了多少/已实现差价/在手浮动 二. 止盈判分 每条自动止盈: 触发过没有; 触发的 = 卖点之后又跌了多少 (保住的钱, 负数=卖飞); 没触发的 = 高水位与武装状态 三. 挂/未挂对照 窗口内 ATTACH 成功的票 vs 想挂但被**每日名额**挡下的票, 各自从决定日到现在的涨跌 —— 名额挡下的是天然对照组 (资格全同, 只差没轮上); 排除项挡下的 (冻结/买不起/黑名单) 结构上就不同, 不进对照。 运行 (桥机 factorevaluation): docker compose run --rm pms-web python scripts/report_strategy_score.py [--days 30] 读数纪律: - 差价用**策略内均价配对** (卖出收益 = 卖出额 − 卖出股数×买入均价), 不做逐笔 FIFO —— 月度读数要的是方向和量级, 不是会计账; 逐笔账在 pms_lot 里, 要精算去查它。 - 现价来自分钟线缓存 (market.get_prices), 收盘后或缓存过期时取不到 → 相应栏目 明说「无现价」, 绝不拿旧价冒充。 - 样本少于三条的段落只列数不下结论 —— 判分要等数据, 不等数据的判分是编故事。 """ import argparse import json import os import sys from datetime import datetime, timedelta sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from app.db.session import fetch_all # noqa: E402 from app.repo import pms_repo # noqa: E402 from app.services.strategy_advisor import ( # noqa: E402 NOTE_AUTO, NOTE_HANDOFF, MARK_HANDOFF_OUT) MIN_SAMPLE = 3 # 少于这个数只报数不下结论 def _f(v, d=0.0): try: return float(v) except (TypeError, ValueError): return d def _pct(a, b): """b 相对 a 的涨跌; 任一边取不到返回 None。""" a, b = _f(a), _f(b) return (b / a - 1.0) if a > 0 and b > 0 else None def _fmt_pct(x): return f"{x:+.2%}" if x is not None else "—" def auto_strategies(): """全部自动策略 (含已归档已撤销), 按 note 前缀识别。""" rows = pms_repo.list_strategies(limit=1000, include_archived=True) return [r for r in rows if str(r.get("note") or "").startswith("自动挂载")] def fills_of(strategy_id): """该策略名下有成交的指令 (exec_qty>0, 不论终态) → 买卖两侧合计。单表。""" rows = fetch_all( "SELECT side, exec_qty, exec_avg_price FROM pms_instruction " "WHERE origin_type = 'strategy' AND origin_id = :sid AND exec_qty > 0", {"sid": strategy_id}) agg = {"buy": {"qty": 0, "amt": 0.0, "n": 0}, "sell": {"qty": 0, "amt": 0.0, "n": 0}} for r in rows: side = str(r.get("side") or "").lower() if side not in agg: continue q, px = int(r.get("exec_qty") or 0), _f(r.get("exec_avg_price")) agg[side]["qty"] += q agg[side]["amt"] += q * px agg[side]["n"] += 1 return agg def ledger_window(days): """窗口内 advisor 留的 ATTACH 痕 (PASS=挂了 / NOTE+名额已满=对照)。单表。""" since = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d %H:%M:%S") rows = fetch_all( "SELECT ts_code, decided_at, verdict, price_at, reason, ref_id " "FROM pms_action_ledger WHERE action = 'ATTACH' AND decided_at >= :since", {"since": since}) attached, control = [], [] for r in rows: if str(r.get("verdict")) == "PASS": attached.append(r) elif "名额已满" in str(r.get("reason") or ""): control.append(r) return attached, control def current_prices(codes): try: from app.services import market return market.get_prices(sorted(set(c for c in codes if c))) except Exception as e: # noqa: BLE001 —— redis 不在 (本地跑) 也要能出报告 print(f" (现价不可用: {type(e).__name__}: {e} —— 浮动与保住栏按无现价处理)") return {} def main(): ap = argparse.ArgumentParser() ap.add_argument("--days", type=int, default=30, help="对照窗口 (自然日, 默认 30)") args = ap.parse_args() print("=" * 66) print("策略判分读数 %s (只读; 窗口 %d 天)" % (datetime.now().strftime("%Y-%m-%d %H:%M:%S"), args.days)) print("=" * 66) try: strats = auto_strategies() except Exception as e: # noqa: BLE001 —— 连不上库也要说人话, 不甩栈 print(f"\n✗ 策略表读取失败: {type(e).__name__}: {e}\n" f" (本脚本要在桥机容器里跑: docker compose run --rm pms-web " f"python scripts/report_strategy_score.py)") sys.exit(1) grids = [s for s in strats if str(s.get("type")).upper() == "GRID"] trails = [s for s in strats if str(s.get("type")).upper() == "TRAIL"] handoff_in = [s for s in strats if str(s.get("note") or "").startswith(NOTE_HANDOFF)] handoff_out = [s for s in strats if MARK_HANDOFF_OUT in str(s.get("note") or "")] codes = {s.get("ts_code") for s in strats} # ---- 零、样本盘点 ---- print("\n【零】样本盘点") if not strats: print(" 还没有任何自动挂载的策略 —— 判分从第一条挂出才开始, 本报告先当空跑冒烟。") else: by_status = {} for s in strats: by_status[s.get("status")] = by_status.get(s.get("status"), 0) + 1 print(f" 自动策略共 {len(strats)} 条 (网格 {len(grids)} / 止盈 {len(trails)}), " f"状态: " + ", ".join(f"{k} {v}" for k, v in sorted(by_status.items()))) print(f" 接力: 换挂出的止盈 {len(handoff_in)} 条, 被接力撤下的网格 {len(handoff_out)} 条") px = current_prices(codes) if strats else {} # ---- 一、网格差价 ---- print("\n【一】网格差价 (含已撤下的; 差价=卖出额−卖出股数×买入均价, 均价配对口径)") if not grids: print(" 无样本。") tot_real, tot_float, n_traded = 0.0, 0.0, 0 for s in grids: agg = fills_of(s["strategy_id"]) b, sl = agg["buy"], agg["sell"] code, sid = s.get("ts_code"), s["strategy_id"] if b["qty"] == 0 and sl["qty"] == 0: print(f" {code} {sid} [{s.get('status')}]: 还没成交过一笔") continue n_traded += 1 buy_avg = b["amt"] / b["qty"] if b["qty"] else 0.0 realized = sl["amt"] - sl["qty"] * buy_avg if sl["qty"] else 0.0 net_qty = b["qty"] - sl["qty"] cur = _f(px.get(code)) floating = net_qty * (cur - buy_avg) if (net_qty > 0 and cur > 0 and buy_avg > 0) else None tot_real += realized if floating is not None: tot_float += floating print(f" {code} {sid} [{s.get('status')}]: " f"买 {b['qty']} 股/{b['amt']:,.0f} 元 (均 {buy_avg:.3f}), " f"卖 {sl['qty']} 股/{sl['amt']:,.0f} 元, 已实现 {realized:+,.0f} 元, " + (f"在手 {net_qty} 股浮动 {floating:+,.0f} 元" if floating is not None else f"在手 {net_qty} 股 (无现价, 浮动不算)")) if grids: print(f" 合计: 有成交的 {n_traded}/{len(grids)} 条, " f"已实现 {tot_real:+,.0f} 元, 可算浮动 {tot_float:+,.0f} 元" + ("" if n_traded >= MIN_SAMPLE else f" (样本不足 {MIN_SAMPLE} 条, 只报数不下结论)")) # ---- 二、止盈判分 ---- print("\n【二】止盈判分 (触发的看「卖点之后又跌了多少」= 保住的钱, 负数=卖飞)") if not trails: print(" 无样本。") n_fired, tot_saved = 0, 0.0 for s in trails: agg = fills_of(s["strategy_id"]) sl = agg["sell"] code, sid = s.get("ts_code"), s["strategy_id"] st = s.get("state") or {} if sl["qty"] > 0: n_fired += 1 sell_avg = sl["amt"] / sl["qty"] cur = _f(px.get(code)) saved = (sell_avg - cur) * sl["qty"] if cur > 0 else None if saved is not None: tot_saved += saved print(f" {code} {sid} [{s.get('status')}]: 触发过, " f"卖 {sl['qty']} 股 (均 {sell_avg:.3f}), " + (f"现价 {cur:.3f} → 保住 {saved:+,.0f} 元" if saved is not None else "无现价, 保住的钱先不算")) else: hw, armed = _f(st.get("high_water")), bool(st.get("armed")) print(f" {code} {sid} [{s.get('status')}]: 未触发, " f"高水位 {hw or '—'}, {'已武装' if armed else '未武装 (垫子还没到武装线)'}") if trails and n_fired: print(f" 合计: 触发 {n_fired}/{len(trails)} 条, 可算的保住 {tot_saved:+,.0f} 元" + ("" if n_fired >= MIN_SAMPLE else f" (样本不足 {MIN_SAMPLE} 条, 只报数不下结论)")) # ---- 三、挂/未挂对照 ---- print(f"\n【三】挂/未挂对照 (近 {args.days} 天; 对照组=资格全同只是没轮上名额的票)") try: attached, control = ledger_window(args.days) except Exception as e: # noqa: BLE001 print(f" ✗ 台账读取失败: {type(e).__name__}: {e}") attached, control = [], [] px3 = current_prices([r.get("ts_code") for r in attached + control]) def _group(rows, label): moves = [] for r in rows: chg = _pct(r.get("price_at"), px3.get(r.get("ts_code"))) d = str(r.get("decided_at"))[:10] print(f" {r.get('ts_code')} {d} 决定价 {_f(r.get('price_at')):.3f} → " f"{_fmt_pct(chg)}") if chg is not None: moves.append(chg) if moves: avg = sum(moves) / len(moves) print(f" {label}: {len(rows)} 次决定, 可算 {len(moves)} 次, 平均 {_fmt_pct(avg)}") else: print(f" {label}: {len(rows)} 次决定, 无一次可算 (决定价或现价缺失)") return moves if not attached and not control: print(" 窗口内没有 ATTACH 留痕 —— 系统还没挂过, 或窗口太短。") else: print(" 挂上的:") m1 = _group(attached, "挂上的") print(" 名额挡下的 (对照):") m2 = _group(control, "对照组") if len(m1) >= MIN_SAMPLE and len(m2) >= MIN_SAMPLE: print(f" 对照差: 挂上的平均 {_fmt_pct(sum(m1) / len(m1))} vs " f"没轮上的 {_fmt_pct(sum(m2) / len(m2))} —— 差值为正说明排序在挑对票") else: print(f" 两组样本 ({len(m1)}/{len(m2)}) 不足 {MIN_SAMPLE}, 本期不下结论。") print("\n判分口径提醒: 本报告只陈述事实读数, 阈值调整 (热度 0.80 / 日上限 2) " "等两周以上样本再议; 精算逐笔账查 pms_lot 与台账。") if __name__ == "__main__": main()