feat: 报告模板系统 - prepare_report_data + generate_report + 5模板
This commit is contained in:
知微
@@ -0,0 +1,127 @@
|
||||
#!/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"
|
||||
|
||||
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()
|
||||
@@ -0,0 +1,139 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
prepare_report_data.py — 为所有报告模板提供结构化参数值
|
||||
输出JSON，每个字段有来源标注，LLM只读不修改
|
||||
用法: python3 prepare_report_data.py [--report-type intraday|strategy|self-buy]
|
||||
"""
|
||||
|
||||
import sqlite3, json, sys, subprocess, re, os
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
DB = Path("/home/hmo/MoFin/data/mofin.db")
|
||||
HK_RATE_SCRIPT = Path("/home/hmo/MoFin/scripts/hk_rate.py")
|
||||
|
||||
def get_hk_rate() -> float:
|
||||
r = subprocess.run(["python3", str(HK_RATE_SCRIPT)], capture_output=True, text=True, timeout=10)
|
||||
m = re.search(r"[\d.]+$", r.stdout.strip())
|
||||
return float(m.group()) if m else 0.93
|
||||
|
||||
def get_portfolio(db, hk_rate):
|
||||
"""总资产/现金/仓位 — 来源: portfolio_summary + holdings实时计算"""
|
||||
r = db.execute("SELECT cash, frozen_cash FROM portfolio_summary WHERE id=1").fetchone()
|
||||
cash, frozen = (r["cash"] or 0), (r["frozen_cash"] or 0)
|
||||
|
||||
total_hkd_mv = 0.0
|
||||
total_cny_mv = 0.0
|
||||
holdings = []
|
||||
for r in db.execute("""
|
||||
SELECT code, name, shares, price, currency, market_value, cost
|
||||
FROM holdings WHERE is_active=1 AND (shares>0 OR shares IS NULL)
|
||||
ORDER BY currency, code
|
||||
"""):
|
||||
shares = r["shares"] or 0
|
||||
if shares == 0:
|
||||
continue
|
||||
cur = r["currency"] or "CNY"
|
||||
price = r["price"] or 0
|
||||
mv = r["market_value"] or (shares * price)
|
||||
cost_total = (r["cost"] or 0) * shares
|
||||
pnl = mv - cost_total
|
||||
pnl_pct = ((price / (r["cost"] or price)) - 1) * 100 if r["cost"] and r["cost"] > 0 else 0
|
||||
|
||||
mv_cny = mv * hk_rate if cur == "HKD" else mv
|
||||
if cur == "HKD":
|
||||
total_hkd_mv += mv
|
||||
else:
|
||||
total_cny_mv += mv_cny
|
||||
|
||||
holdings.append({
|
||||
"code": r["code"],
|
||||
"name": r["name"],
|
||||
"shares": shares,
|
||||
"price": round(price, 2),
|
||||
"currency": cur,
|
||||
"price_display": f"HK${price:.2f}" if cur == "HKD" else f"CNY${price:.2f}",
|
||||
"mv": round(mv, 2),
|
||||
"mv_display": f"HK${mv:.2f}" if cur == "HKD" else f"CNY${mv:.2f}",
|
||||
"pnl_pct": round(pnl_pct, 2),
|
||||
"pnl_amount": round(pnl, 2),
|
||||
})
|
||||
|
||||
stock_value_cny = total_cny_mv + total_hkd_mv * hk_rate
|
||||
total_cash = cash + frozen
|
||||
total_assets = stock_value_cny + total_cash
|
||||
position_pct = (stock_value_cny / total_assets * 100) if total_assets > 0 else 0
|
||||
|
||||
return {
|
||||
"cash": round(cash, 0),
|
||||
"frozen_cash": round(frozen, 0),
|
||||
"total_cash": round(total_cash, 0),
|
||||
"stock_value_cny": round(stock_value_cny, 2),
|
||||
"total_assets": round(total_assets, 2),
|
||||
"position_pct": round(position_pct, 1),
|
||||
"currency": "CNY",
|
||||
"hk_rate": hk_rate,
|
||||
"holdings": holdings,
|
||||
"holdings_count": len(holdings),
|
||||
"source": "portfolio_summary + holdings实时计算",
|
||||
"updated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||||
}
|
||||
|
||||
def get_market(db):
|
||||
"""大盘数据 — 来源: market_snapshots"""
|
||||
r = db.execute("SELECT * FROM market_snapshots ORDER BY id DESC LIMIT 1").fetchone()
|
||||
if not r:
|
||||
return {"error": "无市场数据"}
|
||||
d = dict(r)
|
||||
return {
|
||||
"sh_index": d.get("sh_index", 0),
|
||||
"sz_index": d.get("sz_index", 0),
|
||||
"sh_change": d.get("sh_change_pct", 0),
|
||||
"sz_change": d.get("sz_change_pct", 0),
|
||||
"turnover": d.get("turnover", 0),
|
||||
"advance_decline_ratio": d.get("advance_decline_ratio", 0),
|
||||
"mood": d.get("mood", "unknown"),
|
||||
"snapshot_time": d.get("snapshot_time", ""),
|
||||
"source": "market_snapshots DB",
|
||||
}
|
||||
|
||||
def get_cash_history(db):
|
||||
"""现金变动历史 — 来源: cash_log"""
|
||||
logs = []
|
||||
for r in db.execute("SELECT * FROM cash_log ORDER BY id DESC LIMIT 5"):
|
||||
logs.append({
|
||||
"id": r["id"],
|
||||
"timestamp": r["timestamp"],
|
||||
"cash": r["cash_after"],
|
||||
"change": r["cash_after"] - r["cash_before"],
|
||||
"source": r["source"],
|
||||
"note": r["note"],
|
||||
"verified": bool(r["verified"]),
|
||||
})
|
||||
return logs
|
||||
|
||||
def main():
|
||||
report_type = "intraday"
|
||||
if len(sys.argv) > 1 and sys.argv[1].startswith("--"):
|
||||
report_type = sys.argv[1].split("=")[-1] if "=" in sys.argv[1] else sys.argv[1].lstrip("-")
|
||||
|
||||
db = sqlite3.connect(str(DB))
|
||||
db.row_factory = sqlite3.Row
|
||||
hk_rate = get_hk_rate()
|
||||
|
||||
data = {
|
||||
"_meta": {
|
||||
"report_type": report_type,
|
||||
"generated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"data_integrity": "所有数值由代码从DB获取，LLM不得修改",
|
||||
},
|
||||
"portfolio": get_portfolio(db, hk_rate),
|
||||
"market": get_market(db),
|
||||
"cash_history": get_cash_history(db),
|
||||
}
|
||||
|
||||
print(json.dumps(data, ensure_ascii=False, indent=2))
|
||||
db.close()
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user