chore: deployed pipeline fixes
This commit is contained in:
知微
@@ -1,14 +1,9 @@
|
||||
#!/usr/bin/env python3
|
||||
"""batch_reassess.py — 批量补全12维(九维矩阵)LLM分析（逐只处理，间隔防限流）
|
||||
"""batch_reassess.py — 批量补全九维分析（逐只处理，间隔防限流）
|
||||
|
||||
用法:
|
||||
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 # 单只
|
||||
用法: python3 batch_reassess.py [--all] [--code XXXXXX]
|
||||
|
||||
流程：收集最新数据 → 调LLM(gateway)写12维分析+策略 → 保存到DB
|
||||
流程：收集最新数据 → 调LLM(gateway)写九维分析+策略 → 保存到DB
|
||||
"""
|
||||
import sys, json, subprocess, sqlite3, re, time
|
||||
from datetime import datetime
|
||||
@@ -16,10 +11,9 @@ 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生成的12维分析（>500字）"""
|
||||
"""检查是否为LLM生成的九维分析（>500字）"""
|
||||
conn = sqlite3.connect(DB)
|
||||
r = conn.execute("SELECT LENGTH(full_analysis) FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone()
|
||||
conn.close()
|
||||
@@ -39,34 +33,6 @@ 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}
|
||||
@@ -89,14 +55,7 @@ def collect_data(code):
|
||||
conn.close()
|
||||
|
||||
# 从腾讯API拉最新价和基本面
|
||||
# 代码前缀：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"
|
||||
prefix = "sh" if str(code).startswith(("6","9")) else "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("~")
|
||||
@@ -122,9 +81,8 @@ def collect_data(code):
|
||||
|
||||
def build_prompt(data):
|
||||
"""构建LLM prompt，要求输出完整策略"""
|
||||
cash, total = get_portfolio() # 实时从 portfolio_summary 读
|
||||
if not total:
|
||||
cash, total = 241330, 929727 # 兜底（DB读不到时）
|
||||
cash = 321271 # 可用现金（从DB读取）
|
||||
total = 952879 # 总资产
|
||||
|
||||
# 拉取资金流数据
|
||||
_flow_note = "暂无资金流数据"
|
||||
@@ -313,19 +271,18 @@ def save_result(code, full_text, parsed):
|
||||
|
||||
conn.close()
|
||||
|
||||
def process_stock(code, force_today=False):
|
||||
def process_stock(code):
|
||||
"""处理单只股票"""
|
||||
print(f"\n{'='*50}")
|
||||
print(f"处理: {code}")
|
||||
print(f"{'='*50}")
|
||||
|
||||
if in_cooldown(code):
|
||||
print(f" ⏭ 冷却期内，跳过")
|
||||
if has_llm_analysis(code):
|
||||
print(f" ⏭ 已有LLM九维分析，跳过")
|
||||
return False
|
||||
|
||||
# 有分析且未过期 → 跳过（除非 force_today 且今早未评）
|
||||
if has_llm_analysis(code) and not analysis_stale(code, force_today):
|
||||
print(f" ⏭ 已有12维分析且未过期，跳过")
|
||||
if in_cooldown(code):
|
||||
print(f" ⏭ 冷却期内，跳过")
|
||||
return False
|
||||
|
||||
print(f" 收集数据...", flush=True)
|
||||
@@ -372,41 +329,29 @@ def process_stock(code, force_today=False):
|
||||
|
||||
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)
|
||||
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()
|
||||
rows = conn.execute("SELECT code FROM holding_strategies WHERE status='active' AND decision_type='自选策略' ORDER BY code").fetchall()
|
||||
conn.close()
|
||||
codes = [r[0] for r in rows]
|
||||
|
||||
print(f"待处理: {len(codes)}只 (type={dtype or 'all'}, force_today={force_today})")
|
||||
print(f"待处理: {len(codes)}只")
|
||||
|
||||
ok = 0
|
||||
fail = 0
|
||||
skip = 0
|
||||
for i, code in enumerate(codes):
|
||||
if has_llm_analysis(code) and not analysis_stale(code, force_today):
|
||||
print(f" [{i+1}/{len(codes)}] ⏭ {code} 已有12维分析且未过期")
|
||||
if has_llm_analysis(code):
|
||||
print(f" [{i+1}/{len(codes)}] ⏭ {code} 已有LLM分析")
|
||||
skip += 1
|
||||
continue
|
||||
|
||||
print(f" [{i+1}/{len(codes)}] ", end="", flush=True)
|
||||
if process_stock(code, force_today):
|
||||
if process_stock(code):
|
||||
ok += 1
|
||||
else:
|
||||
fail += 1
|
||||
|
||||
Reference in New Issue
Block a user