fish
230 lines
8.7 KiB
Python
230 lines
8.7 KiB
Python
"""持仓分析 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)
|