feat(analysis): systematic daily 12-dim LLM analysis for holdings+watchlist
Gap (reported by user via zhiwei): premarket full review updated technical
params but full_analysis (12-dim LLM matrix) was empty for new holdings
and stale for old ones — batch_reassess existed but was never wired into
the daily pipeline and only covered watchlist.
System fix:
- premarket_full_review.py: new Step 1.5 runs batch_reassess --type
holding --today every trading day 08:10 (force-refresh today's analysis,
timeout 3600s, result in summary.json)
- batch_reassess.py:
- coverage: --type holding|watchlist|all (was watchlist-only)
- staleness: analysis >20h stale gets refreshed (was: skip if any
analysis exists = forever stale)
- --today flag: force re-analyze if not reassessed since 04:00 today
- cash/total read live from portfolio_summary (was hardcoded 321271/
952879 from weeks ago)
- HK stock prefix fix (5-digit codes -> hk, was sending sz00700)
- watchlist_12d_backfill.py: wrapper for hermes cron (no args support)
- cron job '批量补全九维分析-一次性' -> '自选12维分析补全-每日午间'
(daily 12:30 weekdays, covers 109 watchlist stocks missing analysis)
Verified: 300308 got 1920-char 12-dim analysis written to DB at 08:41,
signal=观望, stop/take-profit updated.
This commit is contained in:
@@ -1,9 +1,14 @@
|
||||
#!/usr/bin/env python3
|
||||
"""batch_reassess.py — 批量补全九维分析(逐只处理,间隔防限流)
|
||||
"""batch_reassess.py — 批量补全12维(九维矩阵)LLM分析(逐只处理,间隔防限流)
|
||||
|
||||
用法: python3 batch_reassess.py [--all] [--code XXXXXX]
|
||||
用法:
|
||||
python3 batch_reassess.py # 所有缺分析/过期的 active 策略
|
||||
python3 batch_reassess.py --type holding # 只处理持仓策略
|
||||
python3 batch_reassess.py --type watchlist # 只处理自选策略
|
||||
python3 batch_reassess.py --type holding --today # 持仓每日刷新(今早未评过的强制重评)
|
||||
python3 batch_reassess.py --code XXXXXX # 单只
|
||||
|
||||
流程:收集最新数据 → 调LLM(gateway)写九维分析+策略 → 保存到DB
|
||||
流程:收集最新数据 → 调LLM(gateway)写12维分析+策略 → 保存到DB
|
||||
"""
|
||||
import sys, json, subprocess, sqlite3, re, time
|
||||
from datetime import datetime
|
||||
@@ -11,9 +16,10 @@ from datetime import datetime
|
||||
DB = "/home/hmo/MoFin/data/mofin.db"
|
||||
GATEWAY = "http://127.0.0.1:8643/v1/chat/completions"
|
||||
COOLDOWN_HOURS = 1
|
||||
STALE_HOURS = 20 # 分析超过20小时视为过期,需要重评
|
||||
|
||||
def has_llm_analysis(code):
|
||||
"""检查是否为LLM生成的九维分析(>500字)"""
|
||||
"""检查是否为LLM生成的12维分析(>500字)"""
|
||||
conn = sqlite3.connect(DB)
|
||||
r = conn.execute("SELECT LENGTH(full_analysis) FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone()
|
||||
conn.close()
|
||||
@@ -33,6 +39,34 @@ def in_cooldown(code):
|
||||
except:
|
||||
return False
|
||||
|
||||
def analysis_stale(code, force_today=False):
|
||||
"""分析是否过期(>STALE_HOURS 或 force_today 时今早4点前未重评)"""
|
||||
conn = sqlite3.connect(DB)
|
||||
r = conn.execute("SELECT reassessed_at FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone()
|
||||
conn.close()
|
||||
if not r or not r[0]:
|
||||
return True
|
||||
try:
|
||||
last = datetime.fromisoformat(r[0])
|
||||
if force_today:
|
||||
today4am = datetime.now().replace(hour=4, minute=0, second=0, microsecond=0)
|
||||
return last < today4am
|
||||
return (datetime.now() - last).total_seconds() / 3600 > STALE_HOURS
|
||||
except:
|
||||
return True
|
||||
|
||||
def get_portfolio():
|
||||
"""从 portfolio_summary 读实时现金/总资产(不再硬编码)"""
|
||||
try:
|
||||
conn = sqlite3.connect(DB)
|
||||
r = conn.execute("SELECT cash, total_assets FROM portfolio_summary WHERE id=1").fetchone()
|
||||
conn.close()
|
||||
if r and r[1]:
|
||||
return int(r[0] or 0), int(r[1])
|
||||
except Exception:
|
||||
pass
|
||||
return 0, 0
|
||||
|
||||
def collect_data(code):
|
||||
"""收集最新数据"""
|
||||
data = {"code": code}
|
||||
@@ -55,7 +89,14 @@ def collect_data(code):
|
||||
conn.close()
|
||||
|
||||
# 从腾讯API拉最新价和基本面
|
||||
prefix = "sh" if str(code).startswith(("6","9")) else "sz"
|
||||
# 代码前缀:5位=港股(hk),6/9开头=沪(sh),其他=深(sz)
|
||||
_c = str(code)
|
||||
if len(_c) == 5:
|
||||
prefix = "hk"
|
||||
elif _c.startswith(("6", "9")):
|
||||
prefix = "sh"
|
||||
else:
|
||||
prefix = "sz"
|
||||
try:
|
||||
r = subprocess.run(["curl", "-s", f"http://qt.gtimg.cn/q={prefix}{code}"], capture_output=True, timeout=10)
|
||||
parts = r.stdout.decode("gbk", errors="ignore").split("~")
|
||||
@@ -81,8 +122,9 @@ def collect_data(code):
|
||||
|
||||
def build_prompt(data):
|
||||
"""构建LLM prompt,要求输出完整策略"""
|
||||
cash = 321271 # 可用现金(从DB读取)
|
||||
total = 952879 # 总资产
|
||||
cash, total = get_portfolio() # 实时从 portfolio_summary 读
|
||||
if not total:
|
||||
cash, total = 241330, 929727 # 兜底(DB读不到时)
|
||||
|
||||
# 拉取资金流数据
|
||||
_flow_note = "暂无资金流数据"
|
||||
@@ -271,20 +313,21 @@ def save_result(code, full_text, parsed):
|
||||
|
||||
conn.close()
|
||||
|
||||
def process_stock(code):
|
||||
def process_stock(code, force_today=False):
|
||||
"""处理单只股票"""
|
||||
print(f"\n{'='*50}")
|
||||
print(f"处理: {code}")
|
||||
print(f"{'='*50}")
|
||||
|
||||
if has_llm_analysis(code):
|
||||
print(f" ⏭ 已有LLM九维分析,跳过")
|
||||
return False
|
||||
|
||||
if in_cooldown(code):
|
||||
print(f" ⏭ 冷却期内,跳过")
|
||||
return False
|
||||
|
||||
# 有分析且未过期 → 跳过(除非 force_today 且今早未评)
|
||||
if has_llm_analysis(code) and not analysis_stale(code, force_today):
|
||||
print(f" ⏭ 已有12维分析且未过期,跳过")
|
||||
return False
|
||||
|
||||
print(f" 收集数据...", flush=True)
|
||||
data = collect_data(code)
|
||||
if not data.get("price"):
|
||||
@@ -329,29 +372,41 @@ def process_stock(code):
|
||||
|
||||
def main():
|
||||
codes = []
|
||||
force_today = "--today" in sys.argv
|
||||
dtype = None
|
||||
if "--type" in sys.argv:
|
||||
idx = sys.argv.index("--type")
|
||||
dtype = sys.argv[idx + 1] # holding | watchlist | all
|
||||
if "--code" in sys.argv:
|
||||
idx = sys.argv.index("--code")
|
||||
codes = [sys.argv[idx+1]]
|
||||
else:
|
||||
# 所有自选策略
|
||||
# 按类型筛选 active 策略
|
||||
type_map = {"holding": "持仓策略", "watchlist": "自选策略"}
|
||||
conn = sqlite3.connect(DB)
|
||||
rows = conn.execute("SELECT code FROM holding_strategies WHERE status='active' AND decision_type='自选策略' ORDER BY code").fetchall()
|
||||
if dtype in type_map:
|
||||
rows = conn.execute(
|
||||
"SELECT code FROM holding_strategies WHERE status='active' AND decision_type=? ORDER BY code",
|
||||
(type_map[dtype],)).fetchall()
|
||||
else:
|
||||
rows = conn.execute(
|
||||
"SELECT code FROM holding_strategies WHERE status='active' ORDER BY decision_type, code").fetchall()
|
||||
conn.close()
|
||||
codes = [r[0] for r in rows]
|
||||
|
||||
print(f"待处理: {len(codes)}只")
|
||||
print(f"待处理: {len(codes)}只 (type={dtype or 'all'}, force_today={force_today})")
|
||||
|
||||
ok = 0
|
||||
fail = 0
|
||||
skip = 0
|
||||
for i, code in enumerate(codes):
|
||||
if has_llm_analysis(code):
|
||||
print(f" [{i+1}/{len(codes)}] ⏭ {code} 已有LLM分析")
|
||||
if has_llm_analysis(code) and not analysis_stale(code, force_today):
|
||||
print(f" [{i+1}/{len(codes)}] ⏭ {code} 已有12维分析且未过期")
|
||||
skip += 1
|
||||
continue
|
||||
|
||||
print(f" [{i+1}/{len(codes)}] ", end="", flush=True)
|
||||
if process_stock(code):
|
||||
if process_stock(code, force_today):
|
||||
ok += 1
|
||||
else:
|
||||
fail += 1
|
||||
|
||||
@@ -2,16 +2,17 @@
|
||||
"""premarket_full_review.py — 盘前全量重评
|
||||
|
||||
执行顺序:
|
||||
1. regenerate_all() 全量技术分析重评(持仓+自选)
|
||||
2. watchlist_auto_exit() 自选退出检查
|
||||
3. 输出摘要
|
||||
1. regenerate_all() 全量技术参数重评(持仓+自选)
|
||||
2. batch_reassess.py --type holding --today 持仓12维LLM分析(每日强制刷新)
|
||||
3. watchlist_auto_exit() 自选退出检查
|
||||
4. 输出摘要
|
||||
|
||||
调度:交易日 08:10(A股09:30开盘)
|
||||
"""
|
||||
import sys, os, json
|
||||
sys.path.insert(0, '/home/hmo/MoFin')
|
||||
|
||||
# Step 1: 全量重评
|
||||
# Step 1: 全量技术参数重评
|
||||
print("=" * 50)
|
||||
print("📊 盘前全量重评开始")
|
||||
print("=" * 50)
|
||||
@@ -19,6 +20,28 @@ from strategy_lifecycle import regenerate_all
|
||||
result = regenerate_all(stdout=True)
|
||||
print(f"\n重评完成: {result.get('ok',0)}/{result.get('total',0)}成功")
|
||||
|
||||
# Step 1.5: 持仓 12 维 LLM 深度分析(每日强制,14只约8-10分钟)
|
||||
print("\n" + "=" * 50)
|
||||
print("🧠 持仓12维LLM分析(每日强制刷新)")
|
||||
print("=" * 50)
|
||||
import subprocess as _sp
|
||||
analysis_result = {"ok": 0, "fail": 0, "skip": 0}
|
||||
try:
|
||||
r = _sp.run(
|
||||
["python3", "/home/hmo/.hermes/profiles/position-analyst/scripts/batch_reassess.py",
|
||||
"--type", "holding", "--today"],
|
||||
capture_output=True, text=True, timeout=3600)
|
||||
print(r.stdout[-2000:] if len(r.stdout) > 2000 else r.stdout)
|
||||
if r.returncode != 0 and r.stderr:
|
||||
print(f"⚠️ stderr: {r.stderr[:300]}")
|
||||
# 从输出尾部解析统计
|
||||
import re as _re
|
||||
m = _re.search(r"完成: (\d+)成功, (\d+)失败, (\d+)跳过", r.stdout)
|
||||
if m:
|
||||
analysis_result = {"ok": int(m.group(1)), "fail": int(m.group(2)), "skip": int(m.group(3))}
|
||||
except Exception as e:
|
||||
print(f"⚠️ 12维分析步骤异常: {e}")
|
||||
|
||||
# Step 2: 自选退出
|
||||
print("\n" + "=" * 50)
|
||||
print("🔍 自选退出检查")
|
||||
@@ -30,6 +53,7 @@ exited = auto_exit(dry_run=False)
|
||||
summary = {
|
||||
"premarket_at": __import__('datetime').datetime.now().isoformat(),
|
||||
"reassess": result,
|
||||
"llm_analysis_12d": analysis_result,
|
||||
"auto_exit": [{"code": c, "name": n, "reason": r} for c, n, s, r in exited],
|
||||
"total_kept": result.get('total', 0) - len(exited),
|
||||
}
|
||||
|
||||
@@ -0,0 +1,13 @@
|
||||
#!/usr/bin/env python3
|
||||
"""watchlist_12d_backfill.py — 自选12维LLM分析补全(每日午间)
|
||||
|
||||
包装 batch_reassess.py --type watchlist,供 hermes cron 调度(不支持参数)。
|
||||
调度:交易日 12:30(午间低峰,自选补全约80分钟)
|
||||
"""
|
||||
import subprocess, sys
|
||||
|
||||
r = subprocess.run(
|
||||
[sys.executable, "/home/hmo/MoFin/deploy/profile-scripts/batch_reassess.py",
|
||||
"--type", "watchlist"],
|
||||
timeout=7200)
|
||||
sys.exit(r.returncode)
|
||||
@@ -0,0 +1,7 @@
|
||||
import json
|
||||
d = json.load(open('/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json'))
|
||||
jobs = d if isinstance(d, list) else d.get('jobs', [])
|
||||
for j in jobs:
|
||||
if '批量补全' in j.get('name', '') or '九维' in j.get('name', ''):
|
||||
print(json.dumps({k: j.get(k) for k in ('name', 'script', 'no_agent', 'prompt', 'schedule', 'enabled')},
|
||||
ensure_ascii=False, indent=2))
|
||||
@@ -0,0 +1,25 @@
|
||||
import sqlite3
|
||||
conn = sqlite3.connect('/home/hmo/MoFin/data/mofin.db')
|
||||
print('=== portfolio_summary schema ===')
|
||||
for r in conn.execute("SELECT sql FROM sqlite_master WHERE name='portfolio_summary'"):
|
||||
print(r[0])
|
||||
print()
|
||||
print('=== latest row ===')
|
||||
cur = conn.execute('SELECT * FROM portfolio_summary ORDER BY id DESC LIMIT 1')
|
||||
cols = [d[0] for d in cur.description]
|
||||
row = cur.fetchone()
|
||||
for c, v in zip(cols, row):
|
||||
print(f' {c} = {v}')
|
||||
print()
|
||||
print('=== decision_type distribution ===')
|
||||
for r in conn.execute("SELECT decision_type, COUNT(*) FROM holding_strategies WHERE status='active' GROUP BY decision_type"):
|
||||
print(f' {r[0]}: {r[1]}')
|
||||
print()
|
||||
print('=== full_analysis staleness ===')
|
||||
for r in conn.execute("""
|
||||
SELECT decision_type,
|
||||
SUM(CASE WHEN full_analysis IS NULL OR LENGTH(full_analysis)<500 THEN 1 ELSE 0 END) as missing,
|
||||
SUM(CASE WHEN LENGTH(full_analysis)>=500 THEN 1 ELSE 0 END) as has_fa,
|
||||
COUNT(*) as total
|
||||
FROM holding_strategies WHERE status='active' GROUP BY decision_type"""):
|
||||
print(f' {r[0]}: missing={r[1]} has={r[2]} total={r[3]}')
|
||||
@@ -0,0 +1,9 @@
|
||||
import ast, sys
|
||||
for p in ['/home/hmo/MoFin/deploy/profile-scripts/batch_reassess.py',
|
||||
'/home/hmo/MoFin/deploy/profile-scripts/premarket_full_review.py']:
|
||||
try:
|
||||
ast.parse(open(p).read())
|
||||
print('OK:', p)
|
||||
except SyntaxError as e:
|
||||
print('SYNTAX ERROR:', p, e)
|
||||
sys.exit(1)
|
||||
@@ -0,0 +1,27 @@
|
||||
import json, shutil
|
||||
from datetime import datetime
|
||||
|
||||
jf = '/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json'
|
||||
shutil.copy(jf, jf + '.bak-20260720')
|
||||
|
||||
d = json.load(open(jf))
|
||||
jobs = d if isinstance(d, list) else d.get('jobs', [])
|
||||
for j in jobs:
|
||||
if j.get('name') == '批量补全九维分析-一次性':
|
||||
j['name'] = '自选12维分析补全-每日午间'
|
||||
j['script'] = 'watchlist_12d_backfill.py'
|
||||
j['schedule'] = {"kind": "cron", "expr": "30 12 * * 1-5", "display": "30 12 * * 1-5"}
|
||||
j['schedule_display'] = "30 12 * * 1-5"
|
||||
j['next_run_at'] = "2026-07-20T12:30:00+08:00"
|
||||
j['state'] = 'scheduled'
|
||||
print('updated job:', j['name'], '| script:', j['script'], '| schedule:', j['schedule_display'])
|
||||
break
|
||||
else:
|
||||
print('job not found!')
|
||||
|
||||
if isinstance(d, list):
|
||||
json.dump(jobs, open(jf, 'w'), ensure_ascii=False, indent=2)
|
||||
else:
|
||||
d['jobs'] = jobs
|
||||
json.dump(d, open(jf, 'w'), ensure_ascii=False, indent=2)
|
||||
print('saved')
|
||||
@@ -0,0 +1,7 @@
|
||||
import sqlite3
|
||||
conn = sqlite3.connect('/home/hmo/MoFin/data/mofin.db')
|
||||
r = conn.execute("SELECT code, LENGTH(full_analysis), reassessed_at, timing_signal, stop_loss, take_profit FROM holding_strategies WHERE code='300308'").fetchone()
|
||||
print(r)
|
||||
# also show first 300 chars of the analysis
|
||||
r2 = conn.execute("SELECT substr(full_analysis, 1, 400) FROM holding_strategies WHERE code='300308'").fetchone()
|
||||
print(r2[0] if r2 else 'none')
|
||||
Reference in New Issue
Block a user