MoFin/deploy/profile-scripts/generate_report.py

#!/usr/bin/env python3
"""
generate_report.py — 模板报告生成器 v2
流程: 代码采集数据 → 构建所有数据段 → 填入模板 → 输出预填模板+参数表
用法: python3 generate_report.py <report_type>

LLM只写分析文本，所有数字由代码保障。
"""

import sys, json, subprocess, re
from pathlib import Path

SCRIPTS_DIR = Path(__file__).parent
TEMPLATES_DIR = SCRIPTS_DIR.parent / "templates"

def get_report_data():
    r = subprocess.run(["python3", str(SCRIPTS_DIR / "prepare_report_data.py")], capture_output=True, text=True, timeout=30)
    if r.returncode != 0:
        print(f"ERROR: {r.stderr}", file=sys.stderr)
        sys.exit(1)
    return json.loads(r.stdout)

def fill_template(text: str, data: dict) -> str:
    def replacer(m):
        full_key = m.group(1)
        # 分离key和format spec
        if ":" in full_key:
            key, fmt = full_key.split(":", 1)
        else:
            key, fmt = full_key, ""
        parts = key.split(".")
        val = data
        try:
            for p in parts:
                val = val[p]
            if val is None:
                return "N/A"
            if fmt:
                try:
                    return format(val, fmt)
                except:
                    return str(val)
            if isinstance(val, float):
                return f"{val:.2f}" if abs(val) < 10000 else f"{val:.0f}"
            return str(val)
        except (KeyError, TypeError):
            return f"【缺失:{full_key}】"
    return re.sub(r'\{([^}]+)\}', replacer, text)

def build_holdings_table(data) -> str:
    """生成持仓明细表"""
    h = data["portfolio"]["holdings"]
    if not h:
        return "（空仓）"
    lines = []
    for stk in h:
        lines.append(f"  {stk['code']} {stk['name']} {stk['shares']}股 {stk['price_display']} {stk['mv_display']} {stk['pnl_pct']:+.2f}%")
    return "\n".join(lines)

def build_risk_holdings(data) -> str:
    """浮亏>20%的持仓"""
    h = data["portfolio"]["holdings"]
    risk = [s for s in h if s["pnl_pct"] < -20]
    if not risk:
        return "无"
    lines = []
    for s in risk:
        lines.append(f"🔴 {s['code']} {s['name']} {s['pnl_pct']:+.2f}% {s['price_display']}")
    return "\n".join(lines)

def build_cash_source(data) -> str:
    """现金来源追溯"""
    logs = data.get("cash_history", [])
    if not logs:
        return "无记录"
    latest = logs[0]
    return f"{latest['note']} (验证:{'✅' if latest['verified'] else '❌'})"

def build_sections(data) -> dict:
    """构建所有动态段"""
    return {
        "HOLDINGS_TABLE": build_holdings_table(data),
        "HOLDINGS_RISK": build_risk_holdings(data),
        "CASH_SOURCE": build_cash_source(data),
        "CASH_AMOUNT": f"{data['portfolio']['cash']:.0f}",
        "TOTAL_ASSETS": f"{data['portfolio']['total_assets']:.0f}",
        "POSITION_PCT": f"{data['portfolio']['position_pct']}",
        "STOCK_VALUE": f"{data['portfolio']['stock_value_cny']:.0f}",
        "HOLDINGS_COUNT": str(data['portfolio']['holdings_count']),
        "GENERATED_AT": data['_meta']['generated_at'],
        "HK_RATE": str(data['portfolio']['hk_rate']),
    }

def main():
    report_type = sys.argv[1] if len(sys.argv) > 1 else "intraday_monitor"
    
    # ★ 前置重评：对 holding_strategies 中信号为买入/卖出的股票，先触发重评
    try:
        import sqlite3
        _conn = sqlite3.connect("/home/hmo/web-dashboard/data/mofin.db")
        _actionable = _conn.execute(
            "SELECT hs.code, lp.price, hs.entry_low, hs.entry_high FROM holding_strategies hs "
            "LEFT JOIN live_prices lp ON hs.code = lp.code "
            "WHERE hs.status LIKE '%active%' "
            "AND hs.timing_signal IN ('买入','可买入','可加仓','卖出','止盈')"
        ).fetchall()
        _conn.close()
        MAX_PRE_REASSESS = 5
        for _code, _price, _el, _eh in _actionable[:MAX_PRE_REASSESS]:
            # 价格必须在买入区内或附近（不高于上沿20%），否则不触发重评
            if _price and _el and _eh and _price > 0 and _el > 0 and _eh > 0:
                if _price > _eh * 1.20:
                    print(f"  ⏭️ {_code}: 价{_price}超买入区上沿+{((_price/_eh)-1)*100:.0f}%，跳过重评")
                    continue
            try:
                subprocess.run(
                    ["python3", str(SCRIPTS_DIR / "per_stock_reassess.py"), _code],
                    capture_output=True, timeout=15
                )
            except:
                pass
    except Exception:
        pass
    
    data = get_report_data()
    sections = build_sections(data)
    
    # 读模板
    template_path = TEMPLATES_DIR / f"{report_type}.txt"
    if not template_path.exists():
        print(f"ERROR: 模板 {template_path} 不存在", file=sys.stderr)
        sys.exit(1)
    template = template_path.read_text(encoding="utf-8")
    
    # 先填大段占位符 {SECTION_NAME}（全大写），再填简单占位符 {xxx.yyy}
    filled = template
    for key, val in sections.items():
        filled = filled.replace(f"{{{key}}}", val)
    filled = fill_template(filled, data)
    
    # 输出
    print(filled)
    print()
    # 参数表附在最后供LLM参考
    print("【参 | 代码采集 | LLM不得修改】")
    param_table = {
        "portfolio": {k: data["portfolio"][k] for k in ["total_assets","cash","frozen_cash","stock_value_cny","position_pct","holdings_count","hk_rate"]},
        "market": {k: data["market"][k] for k in ["sh_index","sz_index","sh_change","sz_change","advance_decline_ratio","mood"]},
        "cash_source": sections["CASH_SOURCE"],
        "data_integrity": data["_meta"]["data_integrity"],
    }
    print(json.dumps(param_table, ensure_ascii=False, indent=2))

if __name__ == "__main__":
    main()