新增持仓分析 prompt 生成功能

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

ft-app/app/prompt_builder.py

@@ -0,0 +1,229 @@
+"""持仓分析 prompt 生成器。
+
+根据最近 5 个交易日的主力持仓排名生成结构化分析 prompt，
+按席位属性（家人/大户/其他/噪音）分类汇总，聚焦资金风向变化。
+"""
+
+from collections import defaultdict
+from datetime import date, timedelta
+
+from app.database import SessionLocal
+from app.models import PositionRanking, Contract
+
+
+# ── 席位分类 ──
+FAMILY_NAMES = [
+    "东方财富", "徽商期货", "方正中期", "中信建投", "平安期货",
+]
+WHALE_NAMES = [
+    "永安期货", "中信期货", "国泰君安", "海通期货", "华泰期货",
+    "高盛期货", "乾坤期货", "摩根大通", "中粮期货",
+]
+NOISE_NAMES = ["东证期货"]
+
+TYPE_LABEL = {"volume": "成交量", "long": "多单持仓", "short": "空单持仓"}
+
+
+def _classify(inst: str) -> str:
+    """匹配席位分类，返回 家人/大户/噪音/其他。"""
+    for kw in FAMILY_NAMES:
+        if kw in inst:
+            return "家人"
+    for kw in WHALE_NAMES:
+        if kw in inst:
+            return "大户"
+    for kw in NOISE_NAMES:
+        if kw in inst:
+            return "噪音"
+    return "其他"
+
+
+def _build_legend() -> str:
+    """生成席位分类图例。"""
+    lines = ["## 席位分类", ""]
+    lines.append(f"- **家人席位（散户）**: {'、'.join(FAMILY_NAMES)}")
+    lines.append(f"- **大户席位（产业/套保）**: {'、'.join(WHALE_NAMES)}")
+    lines.append(f"- **噪音席位（忽略）**: {'、'.join(NOISE_NAMES)}")
+    lines.append("- **其他席位**: 未归入以上三类的席位")
+    lines.append("")
+    return "\n".join(lines)
+
+
+def _format_number(n: int) -> str:
+    """数字格式化为万手。"""
+    v = n / 10000
+    sign = "+" if n > 0 else ""
+    return f"{sign}{v:.2f}万"
+
+
+def build_prompt(contract_code: str | None = None) -> str:
+    """生成单个或全部活跃合约的持仓分析 prompt。
+
+    Args:
+        contract_code: 合约代码（如 FG2609），为 None 时分析所有活跃合约。
+
+    Returns:
+        可直接发送给 AI 进行分析的 prompt 文本。
+    """
+    db = SessionLocal()
+    try:
+        if contract_code:
+            codes = [contract_code.upper()]
+        else:
+            codes = [
+                r[0] for r in
+                db.query(Contract.code)
+                .filter(Contract.is_active == True)
+                .order_by(Contract.code)
+                .all()
+            ]
+
+        # 取最近 5 个有持仓数据的交易日
+        all_dates = sorted({
+            r[0] for r in
+            db.query(PositionRanking.date)
+            .filter(PositionRanking.contract_code.in_(codes))
+            .distinct()
+            .all()
+        }, reverse=True)[:5]
+        all_dates.reverse()  # 从旧到新
+
+        if not all_dates:
+            return "无持仓数据"
+
+        # 拉取全部数据
+        rankings = (
+            db.query(PositionRanking)
+            .filter(
+                PositionRanking.contract_code.in_(codes),
+                PositionRanking.date.in_(all_dates),
+            )
+            .order_by(PositionRanking.contract_code, PositionRanking.date, PositionRanking.data_type, PositionRanking.rank)
+            .all()
+        )
+
+        # 按 contract → date → data_type 组织
+        data: dict[str, dict] = defaultdict(lambda: defaultdict(lambda: defaultdict(list)))
+        for r in rankings:
+            data[r.contract_code][r.date][r.data_type].append(r)
+
+        parts = [_build_legend()]
+
+        for code in codes:
+            if code not in data:
+                continue
+            cd = data[code]
+            parts.append(f"## {code}")
+            parts.append("")
+            parts.append(f"分析区间: {min(all_dates)} → {max(all_dates)}（共 {len(all_dates)} 个交易日）")
+            parts.append("")
+
+            # ── 每日明细 ──
+            for d in all_dates:
+                if d not in cd:
+                    continue
+                parts.append(f"### {d} {_weekday_zh(d)}")
+                parts.append("")
+
+                for dtype in ["volume", "long", "short"]:
+                    rows = cd[d].get(dtype, [])
+                    if not rows:
+                        continue
+                    parts.append(f"**{TYPE_LABEL[dtype]} Top 20**")
+                    parts.append("")
+                    parts.append("| 排名 | 席位 | 分类 | 持仓量 | 增减 |")
+                    parts.append("|------|------|------|--------|------|")
+                    for r in rows[:20]:
+                        tag = _classify(r.institution)
+                        if tag == "噪音":
+                            continue  # 跳过东证
+                        parts.append(
+                            f"| {r.rank} | {r.institution} | {tag} | "
+                            f"{_format_number(r.value)} | {_format_number(r.change)} |"
+                        )
+                    parts.append("")
+
+            # ── 每日家人 vs 大户汇总 ──
+            parts.append(f"### {code} 分类资金变化汇总")
+            parts.append("")
+            parts.append("| 日期 | 家人多单增减 | 家人空单增减 | 大户多单增减 | 大户空单增减 | 其他多单增减 | 其他空单增减 |")
+            parts.append("|------|-------------|-------------|-------------|-------------|-------------|-------------|")
+
+            for d in all_dates:
+                if d not in cd:
+                    continue
+                cat_changes = defaultdict(lambda: defaultdict(int))
+                for dtype in ["long", "short"]:
+                    for r in cd[d].get(dtype, []):
+                        tag = _classify(r.institution)
+                        if tag == "噪音":
+                            continue
+                        cat_changes[tag][dtype] += r.change
+
+                parts.append(
+                    f"| {d} "
+                    f"| {_format_number(cat_changes['家人']['long'])} "
+                    f"| {_format_number(cat_changes['家人']['short'])} "
+                    f"| {_format_number(cat_changes['大户']['long'])} "
+                    f"| {_format_number(cat_changes['大户']['short'])} "
+                    f"| {_format_number(cat_changes['其他']['long'])} "
+                    f"| {_format_number(cat_changes['其他']['short'])} |"
+                )
+            parts.append("")
+
+            # ── 5日累计 ──
+            parts.append(f"### {code} 5日累计资金变化")
+            parts.append("")
+            total_cat = defaultdict(lambda: defaultdict(int))
+            for d in all_dates:
+                if d not in cd:
+                    continue
+                for dtype in ["long", "short"]:
+                    for r in cd[d].get(dtype, []):
+                        tag = _classify(r.institution)
+                        if tag == "噪音":
+                            continue
+                        total_cat[tag][dtype] += r.change
+
+            parts.append("| 分类 | 5日多单累计 | 5日空单累计 | 净多（多-空） |")
+            parts.append("|------|------------|------------|--------------|")
+            for tag in ["家人", "大户", "其他"]:
+                net = total_cat[tag]["long"] - total_cat[tag]["short"]
+                net_str = _format_number(net)
+                parts.append(
+                    f"| {tag} | {_format_number(total_cat[tag]['long'])} "
+                    f"| {_format_number(total_cat[tag]['short'])} | {net_str} |"
+                )
+            parts.append("")
+
+        # ── 分析引导 ──
+        parts.append("---")
+        parts.append("")
+        parts.append("请基于以上数据进行分析，重点关注：")
+        parts.append("")
+        parts.append("1. **家人席位动向**: 5日内多空增减趋势，散户情绪偏多还是偏空，"
+                      "是否有明显的追涨杀跌行为。")
+        parts.append("2. **大户席位动向**: 产业资金和套保资金的布局方向，"
+                      "多空增减是否与家人席位形成对手盘。")
+        parts.append("3. **多空力量对比**: 各分类的净多/净空方向，"
+                      "整体市场情绪偏向。")
+        parts.append("4. **异常信号**: 单日大幅增减的席位，"
+                      "分类资金出现明显分歧的交易日。")
+        parts.append("5. **行情预判**: 结合席位资金流向，"
+                      "对下一交易日走势给出偏多/偏空/震荡的判断及理由。")
+
+        return "\n".join(parts)
+    finally:
+        db.close()
+
+
+def _weekday_zh(d: date) -> str:
+    return ["周一", "周二", "周三", "周四", "周五", "周六", "周日"][d.weekday()]
+
+
+if __name__ == "__main__":
+    import sys
+    sys.path.insert(0, "/app")
+    from app.prompt_builder import build_prompt
+    prompt = build_prompt()
+    print(prompt)