tradingSystem/scripts/report_strategy_score.py

# -*- 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
        if sl["qty"] and not b["qty"]:
            # 只有卖出没有买入 (买入腿痕迹缺失/被对账冲销): 按零成本算会把整笔卖出额
            # 报成利润, 判分虚高误导阈值决策 (2026-08-28 修) —— 这条不算差价, 单独点名。
            print(f"  {code} {sid} [{s.get('status')}]: 卖 {sl['qty']} 股/{sl['amt']:,.0f} 元, "
                  f"但**查无买入腿成交** —— 差价无法计算, 不计入合计, 请核对该策略的成交归属")
            continue
        sell_q = min(sl["qty"], b["qty"])
        if sl["qty"] > b["qty"]:
            print(f"  {code} {sid}: 卖出 {sl['qty']} 股 > 买入 {b['qty']} 股, "
                  f"超出部分 {sl['qty'] - b['qty']} 股不属本策略买入, 差价只按配对部分算")
        realized = (sl["amt"] * (sell_q / sl["qty"]) - sell_q * 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()