revert(34337fc5): 恢复被知微二次stale提交覆盖的昨晚重构——batch/per_stock/stale_detector/fix_gateway_port/candidate_filter 回滚至1e71a2d8版本;她提交中的运行时文件(db-shm/db-wal/price_history/market_scan_summary)移出跟踪;保留其morning_health_check小果清理

This commit is contained in:
hmo
2026-07-21 23:24:45 +08:00
parent da4112b430
commit 05a60cf0d4
10 changed files with 332 additions and 2543 deletions
+5
View File
@@ -39,3 +39,8 @@ scripts/mo_data.py
data/prompts/ data/prompts/
data/backups/ data/backups/
data/mofin.db-shm
data/mofin.db-wal
data/price_history.json
deploy/profile-scripts/data/
BIN
View File
Binary file not shown.
View File
File diff suppressed because it is too large Load Diff
+181 -64
View File
@@ -1,19 +1,30 @@
#!/usr/bin/env python3 #!/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 import sys, json, subprocess, sqlite3, re, time, os
from datetime import datetime from datetime import datetime
# ── 共享 LLM 客户端 + DB 工具(profile-scripts 硬链到同目录)──
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, "/home/hmo/MoFin")
from llm_client import call_llm, REASSESS_MODEL, gateway_alive, ocg_alive
from mofin_db import snapshot_strategy_history
DB = "/home/hmo/MoFin/data/mofin.db" DB = "/home/hmo/MoFin/data/mofin.db"
GATEWAY = "http://127.0.0.1:8643/v1/chat/completions"
COOLDOWN_HOURS = 1 COOLDOWN_HOURS = 1
STALE_HOURS = 20 # 分析超过20小时视为过期,需要重评
def has_llm_analysis(code): def has_llm_analysis(code):
"""检查是否为LLM生成的维分析(>500字)""" """检查是否为LLM生成的12维分析(>500字)"""
conn = sqlite3.connect(DB) conn = sqlite3.connect(DB)
r = conn.execute("SELECT LENGTH(full_analysis) FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone() r = conn.execute("SELECT LENGTH(full_analysis) FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone()
conn.close() conn.close()
@@ -33,13 +44,41 @@ def in_cooldown(code):
except: except:
return False 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): def collect_data(code):
"""收集最新数据""" """收集最新数据(含完整策略原文)"""
data = {"code": code} data = {"code": code}
# 从DB读策略 # 从DB读策略(含 full_analysis / changelog_json / position_advice
conn = sqlite3.connect(DB) conn = sqlite3.connect(DB)
r = conn.execute("SELECT name, entry_low, entry_high, stop_loss, take_profit, timing_signal, action, rr_ratio, tech_snapshot, sector_context, stock_category FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone() r = conn.execute("SELECT name, entry_low, entry_high, stop_loss, take_profit, timing_signal, action, rr_ratio, tech_snapshot, sector_context, stock_category, full_analysis, changelog_json, reassessed_at, position_advice FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone()
if r: if r:
data["name"] = r[0] data["name"] = r[0]
data["entry_low"] = r[1] or 0 data["entry_low"] = r[1] or 0
@@ -52,10 +91,21 @@ def collect_data(code):
data["tech_snapshot"] = r[8] or "" data["tech_snapshot"] = r[8] or ""
data["sector_context"] = r[9] or "" data["sector_context"] = r[9] or ""
data["stock_category"] = r[10] or "" data["stock_category"] = r[10] or ""
data["full_analysis"] = r[11] or ""
data["changelog_json"] = r[12] or ""
data["reassessed_at"] = r[13] or ""
data["position_advice"] = r[14] or ""
conn.close() conn.close()
# 从腾讯API拉最新价和基本面 # 从腾讯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: try:
r = subprocess.run(["curl", "-s", f"http://qt.gtimg.cn/q={prefix}{code}"], capture_output=True, timeout=10) 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("~") parts = r.stdout.decode("gbk", errors="ignore").split("~")
@@ -80,9 +130,10 @@ def collect_data(code):
return data return data
def build_prompt(data): def build_prompt(data):
"""构建LLM prompt要求输出完整策略""" """构建LLM prompt先审阅原策略再结合实时数据输出修改判断+九维矩阵分析"""
cash = 321271 # 可用现金(从DB读取) cash, total = get_portfolio()
total = 952879 # 总资产 if not total:
cash, total = 241330, 929727 # 兜底(DB读不到时)
# 拉取资金流数据 # 拉取资金流数据
_flow_note = "暂无资金流数据" _flow_note = "暂无资金流数据"
@@ -122,7 +173,53 @@ def build_prompt(data):
except: except:
pass pass
return f"""你是一个资深A股分析师。请对{data['code']} {data.get('name','')}做一个完整的九维矩阵分析,并输出策略参数。 # ── 构建【原策略全文】section ──
_params_parts = []
if data.get('action'): _params_parts.append(f"当前策略: {data['action']}")
if data.get('timing_signal'): _params_parts.append(f"信号: {data['timing_signal']}")
if data.get('entry_low') or data.get('entry_high'):
_params_parts.append(f"买入区间: {data.get('entry_low',0)}~{data.get('entry_high',0)}")
if data.get('stop_loss'): _params_parts.append(f"止损: {data['stop_loss']}")
if data.get('take_profit'): _params_parts.append(f"止盈: {data['take_profit']}")
if data.get('position_advice'): _params_parts.append(f"仓位: {data['position_advice']}")
_params_str = " | ".join(_params_parts) if _params_parts else "无策略参数"
# 最近3条变更记录
_changelog_str = "无变更记录"
try:
_cl_raw = data.get('changelog_json', '')
if _cl_raw:
_cl = json.loads(_cl_raw) if isinstance(_cl_raw, str) else _cl_raw
if isinstance(_cl, list) and _cl:
_recent = _cl[-3:] if len(_cl) > 3 else _cl
_cl_lines = []
for i, c in enumerate(_recent):
_act = c.get('action', c.get('reason', '')) if isinstance(c, dict) else str(c)
_ts = c.get('timestamp', '') if isinstance(c, dict) else ''
_cl_lines.append(f" {i+1}. {_ts[:16]} {_act[:80]}")
if _cl_lines:
_changelog_str = "\n".join(_cl_lines)
except:
pass
# 完整分析原文(不截断)
_full_analysis = data.get('full_analysis', '') or ''
_fa_display = _full_analysis if _full_analysis else '(首次分析,无历史)'
_orig_strategy_section = f"""当前策略参数: {_params_str}
变更记录(最近3条):
{_changelog_str}
完整分析原文:
{_fa_display}"""
return f"""你是一个资深A股分析师。请先审阅以下【原策略全文】,判断是否需要修改策略,然后做出完整的九维矩阵分析。
【原策略全文】
{_orig_strategy_section}
── 以上是已有的策略,以下是当前实时数据,请结合两者做出判断 ──
⚠️ 重要:以下9个维度不是独立分析的,你必须交叉对比后给出综合结论。 ⚠️ 重要:以下9个维度不是独立分析的,你必须交叉对比后给出综合结论。
例如:如果消息面利好但资金流在流出,说明利好可能是出货;如果基本面强但技术面破位,说明估值可能还没到底。 例如:如果消息面利好但资金流在流出,说明利好可能是出货;如果基本面强但技术面破位,说明估值可能还没到底。
@@ -136,11 +233,18 @@ PE={data.get('pe','?')}(最新财报) 市值={data.get('mcap','?')}亿
资金流:{_flow_note}(近5日累计) 资金流:{_flow_note}(近5日累计)
消息面:{_news_note}(最近3条,自动标注抓取时间) 消息面:{_news_note}(最近3条,自动标注抓取时间)
当前信号:{data.get('timing_signal','?')} 分类:{data.get('stock_category','?')} 当前信号:{data.get('timing_signal','?')} 分类:{data.get('stock_category','?')}
原策略:{(data.get('action','') or '')[:200]}
我的总资产={total}元,可用现金={cash}元。 我的总资产={total}元,可用现金={cash}元。
请严格按以下格式输出: 请严格按以下格式输出(注意节标题不可省略)
【维持或修改】明确二选一判断:维持原策略 / 需要修改策略
【修改点及理由】
如果维持原策略 → 写"无需修改"
如果需要修改 → 逐条列出(每条格式:"- 修改点名称:理由说明"
【最终新策略】
用自然语言输出完整的最终策略全文(200-400字),自包含核心交易逻辑、买入区间价格、止损价、止盈价、仓位比例、风险提示。
⚠️ 本段不要使用【综合结论】【买入区间】等标签——用自然语言描述即可。
【交叉分析】用2-3句话说明哪些维度出现矛盾/共振,最关键的信号是什么 【交叉分析】用2-3句话说明哪些维度出现矛盾/共振,最关键的信号是什么
① 大盘×基本面 [一句话,说明矛盾关系] ① 大盘×基本面 [一句话,说明矛盾关系]
@@ -162,13 +266,18 @@ PE={data.get('pe','?')}(最新财报) 市值={data.get('mcap','?')}亿
【建议止损】数字 【建议止损】数字
【建议止盈】数字 【建议止盈】数字
【建议仓位】只有综合结论"买入"才输出此项。仓位计算公式: 【建议仓位】⚠️不可省略。综合结论"买入""不新建仓";为"买入"时按以下公式:
基础仓位按RR确定:RR<1.5→不推荐,RR1.5~3→8%RR3~5→12%RR5+→15% 基础仓位按RR确定:RR<1.5→不推荐,RR1.5~3→8%RR3~5→12%RR5+→15%
大盘偏弱×0.8,大盘偏强×1.15 大盘偏弱×0.8,大盘偏强×1.15
蓝筹/白马×1.2,成长×0.85,题材/短线×0.6 蓝筹/白马×1.2,成长×0.85,题材/短线×0.6
最终仓位范围:5%~20% 最终仓位范围:5%~20%
同时考虑:现金{cash}元足够买多少手。 同时考虑:现金{cash}元足够买多少手。
输出格式:"X%(理由:一句话说明为什么这个仓位)""" 输出格式:"X%(理由:一句话说明为什么这个仓位)"
⚠️ 输出纪律(必须遵守):
1. 直接以【维持或修改】开头,禁止任何寒暄、开场白、分隔线
2. 禁止输出 <structured_data> 或任何 XML/JSON/代码块
3. 所有【】节标题一个都不能少"""
def parse_response(text): def parse_response(text):
"""从LLM回复中提取策略参数""" """从LLM回复中提取策略参数"""
result = {"signal": "", "entry_low": 0, "entry_high": 0, "stop_loss": 0, "take_profit": 0, "position": ""} result = {"signal": "", "entry_low": 0, "entry_high": 0, "stop_loss": 0, "take_profit": 0, "position": ""}
@@ -217,10 +326,13 @@ def parse_response(text):
return result return result
def save_result(code, full_text, parsed): def save_result(code, full_text, parsed):
"""保存LLM结果到DB""" """保存LLM结果到DB(先快照再UPDATE"""
conn = sqlite3.connect(DB) conn = sqlite3.connect(DB)
now = datetime.now().isoformat() now = datetime.now().isoformat()
# ── 修改前快照 ──
snapshot_strategy_history(conn, code, 'batch_12d')
updates = ["full_analysis=?", "reassessed_at=?"] updates = ["full_analysis=?", "reassessed_at=?"]
params = [full_text, now] params = [full_text, now]
@@ -260,106 +372,111 @@ def save_result(code, full_text, parsed):
_tp = parsed.get("take_profit", 0) _tp = parsed.get("take_profit", 0)
_pos = parsed.get("position", "") _pos = parsed.get("position", "")
_msg = f"📈 {_name}({code}) 价{_p}→12维分析生成买入信号!区间{_el}~{_eh}{_sl}{_tp} 仓位{_pos}" _msg = f"📈 {_name}({code}) 价{_p}→12维分析生成买入信号!区间{_el}~{_eh}{_sl}{_tp} 仓位{_pos}"
import urllib.request, json as _jj from alert_helper import notify as _notify, ACTION as _ACT
_req = urllib.request.Request("http://127.0.0.1:5805/", _notify("买入信号", _msg, _ACT)
data=_jj.dumps({"body": _msg, "to": "hmo@yoin.fun", "type": "chat"}).encode(), print(f" \U0001f4e8 XMPP推送成功: {_msg[:60]}")
headers={"Content-Type": "application/json"})
urllib.request.urlopen(_req, timeout=5)
print(f" 📨 XMPP推送成功: {_msg[:60]}")
except Exception as _e: except Exception as _e:
print(f" ⚠️ XMPP推送失败: {_e}") print(f" \u26a0\ufe0f XMPP推送失败: {_e}")
conn.close() conn.close()
def process_stock(code): def process_stock(code, force_today=False):
"""处理单只股票""" """处理单只股票"""
print(f"\n{'='*50}") print(f"\n{'='*50}")
print(f"处理: {code}") print(f"处理: {code}")
print(f"{'='*50}") print(f"{'='*50}")
if has_llm_analysis(code): if in_cooldown(code):
print(f" ⏭ 已有LLM九维分析,跳过") print(f" \u23ed 冷却期内,跳过")
return False return False
if in_cooldown(code): # 有分析且未过期 \u2192 跳过(除非 force_today 且今早未评)
print(f" ⏭ 冷却期内,跳过") if has_llm_analysis(code) and not analysis_stale(code, force_today):
print(f" \u23ed 已有12维分析且未过期,跳过")
return False return False
print(f" 收集数据...", flush=True) print(f" 收集数据...", flush=True)
data = collect_data(code) data = collect_data(code)
if not data.get("price"): if not data.get("price"):
print(f" ⚠️ 无价格数据,跳过") print(f" \u26a0\ufe0f 无价格数据,跳过")
return False return False
print(f" 调LLM生成九维分析...", flush=True) print(f" 调LLM生成九维分析...", flush=True)
prompt = build_prompt(data) prompt = build_prompt(data)
try: # ── 使用共享 LLM 客户端(替代 curl subprocess)──
r = subprocess.run(["curl", "-s", "--max-time", "300", result = call_llm(prompt, model=REASSESS_MODEL, max_tokens=4096)
"-H", "Content-Type: application/json",
"-H", "Authorization: Bearer hermes123",
"-d", json.dumps({"model":"deepseek-v4-flash","messages":[{"role":"user","content":prompt}],"max_tokens":2048}),
GATEWAY], capture_output=True, timeout=310)
if r.returncode != 0: if not result["ok"]:
print(f" ❌ curl失败: {r.stderr.decode()[:100]}") print(f" \u274c LLM调用失败: {result.get('error','未知错误')}")
return False return False
resp = json.loads(r.stdout) full_text = result["content"]
if "choices" not in resp: print(f" \u2705 LLM返回({len(full_text)}字, {result['elapsed']:.1f}s, 尝试{result['attempts']}次)", flush=True)
print(f" ❌ API异常: {str(resp)[:200]}")
return False
full_text = resp["choices"][0]["message"]["content"]
print(f" ✅ LLM返回({len(full_text)}字)", flush=True)
parsed = parse_response(full_text) parsed = parse_response(full_text)
print(f" 信号={parsed['signal']} 区间={parsed['entry_low']}~{parsed['entry_high']} 损={parsed['stop_loss']} 盈={parsed['take_profit']} 仓位={parsed['position']}") print(f" 信号={parsed['signal']} 区间={parsed['entry_low']}~{parsed['entry_high']} 损={parsed['stop_loss']} 盈={parsed['take_profit']} 仓位={parsed['position']}")
save_result(code, full_text, parsed) save_result(code, full_text, parsed)
print(f" 已保存到DB") print(f" \u2705 已保存到DB")
return True return True
except subprocess.TimeoutExpired:
print(f" ❌ 超时")
return False
except Exception as e:
print(f" ❌ 错误: {e}")
return False
def main(): def main():
# ── 双通道预检:OCG直连 + hermes gateway 兜底,全挂才退出 ──
_ocg_ok = ocg_alive()
_gw_ok = gateway_alive()
if not _ocg_ok and not _gw_ok:
print("[FATAL] OCG上游与hermes gateway均不可用,退出")
sys.exit(1)
if not _ocg_ok:
print("[WARN] OCG直连不可用,将使用gateway兜底(agent运行时,较慢)")
if not _gw_ok:
print("[WARN] hermes gateway不可用,仅使用OCG直连")
codes = [] 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: if "--code" in sys.argv:
idx = sys.argv.index("--code") idx = sys.argv.index("--code")
codes = [sys.argv[idx+1]] codes = [sys.argv[idx+1]]
else: else:
# 所有自选策略 # 按类型筛选 active 策略
type_map = {"holding": "持仓策略", "watchlist": "自选策略"}
conn = sqlite3.connect(DB) 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() conn.close()
codes = [r[0] for r in rows] 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 ok = 0
fail = 0 fail = 0
skip = 0 skip = 0
for i, code in enumerate(codes): for i, code in enumerate(codes):
if has_llm_analysis(code): if has_llm_analysis(code) and not analysis_stale(code, force_today):
print(f" [{i+1}/{len(codes)}] {code} 已有LLM分析") print(f" [{i+1}/{len(codes)}] \u23ed {code} 已有12维分析且未过期")
skip += 1 skip += 1
continue continue
print(f" [{i+1}/{len(codes)}] ", end="", flush=True) print(f" [{i+1}/{len(codes)}] ", end="", flush=True)
if process_stock(code): if process_stock(code, force_today):
ok += 1 ok += 1
else: else:
fail += 1 fail += 1
# 间隔15秒(防gateway过载 # 间隔8秒(pro model较重但gateway可承受;retry逻辑吸收瞬断
if i < len(codes) - 1: if i < len(codes) - 1:
print(f" 等待15秒...", flush=True) print(f" 等待8秒...", flush=True)
time.sleep(15) time.sleep(8)
print(f"\n{'='*50}") print(f"\n{'='*50}")
print(f"完成: {ok}成功, {fail}失败, {skip}跳过") print(f"完成: {ok}成功, {fail}失败, {skip}跳过")
+3 -1
View File
@@ -17,7 +17,9 @@ DB_PATH = Path("/home/hmo/MoFin/data/mofin.db")
UA = "Mozilla/5.0" UA = "Mozilla/5.0"
def get_conn(): def get_conn():
return sqlite3.connect(str(DB_PATH)) c = sqlite3.connect(str(DB_PATH), timeout=30)
c.execute("PRAGMA busy_timeout=30000")
return c
def log_candidate(conn, code, stage, passed, detail): def log_candidate(conn, code, stage, passed, detail):
"""记录过滤日志""" """记录过滤日志"""
@@ -1,7 +0,0 @@
{
"timestamp": "2026-07-21 11:30",
"source": "ths",
"sector_count": 90,
"xiaoguo_status": "offline",
"note": "小果不在线,未做LLM全市场筛选"
}
+16 -23
View File
@@ -21,33 +21,26 @@ def port_open(port, host="127.0.0.1"):
s.close() s.close()
def check_session_health(): def check_session_health():
"""gateway API,检测session是否卡死。超过15s无响应→不健康""" """检测 gateway LLM 是否可用——扫 agent.log 最近一次真实调用结果。
不再发真实 LLM ping25s 超时对 20-100s 的冷启动延迟必误报,且每次白烧 22k token)。
"""
try: try:
payload = json.dumps({ sys.path.insert(0, '/home/hmo/MoFin')
"model": "hermes-agent", from xmpp_logger import _scan_agent_log
"messages": [{"role": "user", "content": "ping"}] r = _scan_agent_log(time.time(), "zhiwei")
}).encode() if r["status"] == "ok":
req = urllib.request.Request(GATEWAY_URL, data=payload, method="POST") print(f"Session {SESSION_ID} 健康 ✓ (agent.log: latency={r.get('latency')}, {r.get('age_sec')}s前)")
req.add_header("Content-Type", "application/json")
req.add_header("Authorization", f"Bearer {API_KEY}")
req.add_header("X-Hermes-Session-Id", SESSION_ID)
t0 = time.time()
with urllib.request.urlopen(req, timeout=25) as r:
data = json.loads(r.read())
reply = data.get("choices", [{}])[0].get("message", {}).get("content", "")
elapsed = time.time() - t0
if reply:
print(f"Session {SESSION_ID} 健康 ✓ ({elapsed:.1f}s)")
return True return True
else: # error/unknown:只有近期有明确失败记录才判不健康
print(f"Session {SESSION_ID} 返回空", file=sys.stderr) if r["status"] == "error":
return False print(f"Session {SESSION_ID} 不健康: agent.log 最近调用失败 — {r.get('error','')[:100]}", file=sys.stderr)
except urllib.request.HTTPError as e:
print(f"Session {SESSION_ID} HTTP错误: {e.code}", file=sys.stderr)
return False return False
# unknown(无近期调用记录)= 空闲,不算不健康
print(f"Session {SESSION_ID} 无近期调用记录(空闲正常)")
return True
except Exception as e: except Exception as e:
print(f"Session {SESSION_ID} 健康: {e}", file=sys.stderr) print(f"Session {SESSION_ID} 健康检查异常: {e}(按健康处理)", file=sys.stderr)
return False return True
def restart_gateway(): def restart_gateway():
"""通过systemd重启gateway""" """通过systemd重启gateway"""
+101 -26
View File
@@ -34,8 +34,11 @@ def _in_cooldown(code):
sys.path.insert(0, "/home/hmo/web-dashboard") sys.path.insert(0, "/home/hmo/web-dashboard")
sys.path.insert(0, "/home/hmo/MoFin") sys.path.insert(0, "/home/hmo/MoFin")
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) # profile-scripts 硬链目录
from strategy_lifecycle import reassess_with_context as reassess_strategy from strategy_lifecycle import reassess_with_context as reassess_strategy
from mo_data import read_decisions, read_portfolio from mo_data import read_decisions, read_portfolio
from llm_client import call_llm, REASSESS_MODEL
from mofin_db import snapshot_strategy_history
def _build_full_analysis(code, entry, result): def _build_full_analysis(code, entry, result):
@@ -431,18 +434,76 @@ def main():
except: except:
pass pass
_prompt = f"""你是一个资深股票分析师。请对股票{code}做一个完整的12维矩阵分析(3横×4纵:大盘/行业/个股 × 基本面/消息面/技术面/资金面)。 # ── 拉取已有策略全文 + 最近变更 ──
_existing_full_analysis = ""
_existing_changelog_text = "无变更记录"
try:
_edb = __import__('sqlite3').connect("/home/hmo/MoFin/data/mofin.db")
_er = _edb.execute(
"SELECT full_analysis, changelog_json FROM holding_strategies "
"WHERE code=? AND status='active'", (code,)
).fetchone()
if _er:
_existing_full_analysis = _er[0] or ""
_cl_raw = _er[1] or ""
if _cl_raw:
_cl = __import__('json').loads(_cl_raw) if isinstance(_cl_raw, str) else _cl_raw
if isinstance(_cl, list) and _cl:
_recent = _cl[-3:]
_existing_changelog_text = "\n".join(
[f" [{c.get('timestamp','?')}] {c.get('action','?')}: {c.get('reason','')}"[:120]
for c in reversed(_recent)]
)
_edb.close()
except:
pass
_prompt = f"""你是一个资深股票分析师。请对股票{code}评估现有策略是否仍然有效,并输出完整的新策略。
╔══════════════════════════════════════════════╗
║ 📋 第一步:审阅原策略 ║
╚══════════════════════════════════════════════╝
【原策略全文】(上次完整分析):
{_existing_full_analysis or '暂无完整策略分析'}
【当前策略参数】:
价格={price} 信号={result.get("timing_signal") or entry.get("timing_signal","")}
买入区间={entry.get("entry_low",0)}~{entry.get("entry_high",0)}
止损={entry.get("stop_loss",0)} 止盈={entry.get("take_profit",0)}
RR={result.get("rr_ratio", entry.get("rr_ratio", 0))}
策略={result.get("action") or entry.get("action","")}
行业={(result.get("sector_context") or entry.get("sector_context",""))[:50]}(当日实时)
技术={(result.get("tech_snapshot") or entry.get("tech_snapshot",""))[:200]}MA=5/10/20/60日 支撑阻力=近20日 量价=当日+近5日趋势)
【最近变更记录】:
{_existing_changelog_text}
╔══════════════════════════════════════════════╗
║ 📊 第二步:12维矩阵交叉分析 ║
╚══════════════════════════════════════════════╝
⚠️ 重要:12个维度必须交叉对比,找出矛盾/共振点,给出综合判断。 ⚠️ 重要:12个维度必须交叉对比,找出矛盾/共振点,给出综合判断。
当前数据(实时API每条标注时间窗口,禁止使用模型训练数据): 当前实时数据(每条标注时间窗口,禁止使用模型训练数据):
大盘={_macro_desc or "震荡"}(当日实时) | PE/市值={_pe_val} {_pb_val}(最新财报) | 价格={price} 区间={entry.get("entry_low",0)}~{entry.get("entry_high",0)} 止损={entry.get("stop_loss",0)} 止盈={entry.get("take_profit",0)} RR={result.get("rr_ratio",entry.get("rr_ratio",0))} | 信号={result.get("timing_signal") or entry.get("timing_signal","")} | 行业={(result.get("sector_context") or entry.get("sector_context",""))[:50]}(当日实时) 大盘={_macro_desc or "震荡"}(当日实时) | PE/市值={_pe_val} {_pb_val}(最新财报)
策略={(result.get("action") or entry.get("action",""))[:200]}
技术={(result.get("tech_snapshot") or entry.get("tech_snapshot",""))[:200]}MA=5/10/20/60日 支撑阻力=近20日 量价=当日+近5日趋势)
资金流={_flow_note}(近5日累计) 资金流={_flow_note}(近5日累计)
消息面={_news_note}(最近3条,自动标注抓取时间) 消息面={_news_note}(最近3条,自动标注抓取时间)
格式: ╔══════════════════════════════════════════════╗
║ 📝 第三步:决策输出 ║
╚══════════════════════════════════════════════╝
请严格按以下顺序输出:
【维持或修改】判断当前策略是否仍然有效,回答「维持」或「修改」。
【修改点及理由】(如果维持,写「无需修改」;如果修改,逐条列出):
- 修改什么参数/方向
- 理由(引用具体维度矛盾或共振)
【最终新策略】(完整策略全文,self-contained,可直接存入DB
【交叉分析】哪些维度矛盾/共振,关键信号 【交叉分析】哪些维度矛盾/共振,关键信号
① 大盘×基本面 ② 大盘×消息面 ③ 大盘×技术面 ④ 大盘×资金面 ① 大盘×基本面 ② 大盘×消息面 ③ 大盘×技术面 ④ 大盘×资金面
⑤ 行业×基本面 ⑥ 行业×消息面 ⑦ 行业×技术面 ⑧ 行业×资金面 ⑤ 行业×基本面 ⑥ 行业×消息面 ⑦ 行业×技术面 ⑧ 行业×资金面
@@ -452,26 +513,42 @@ def main():
【综合结论】(买入/关注/观望/卖出) 【综合结论】(买入/关注/观望/卖出)
【操作建议】 【操作建议】
【建议止损】 【建议止损】
【建议止盈】""" 【建议止盈】
try: 【建议仓位】⚠️不可省略,非"买入"时写"不新建仓"
_ur = __import__('urllib.request', fromlist=['Request'])
_req = _ur.Request("http://127.0.0.1:8643/v1/chat/completions",
data=__import__('json').dumps({"model":"deepseek-v4-flash","messages":[{"role":"user","content":_prompt}],"max_tokens":1024}).encode(),
headers={"Content-Type":"application/json","Authorization":"Bearer hermes123"})
_resp = _ur.build_opener(_ur.ProxyHandler({})).open(_req, timeout=300)
_llm_out = __import__('json').loads(_resp.read().decode())["choices"][0]["message"]["content"]
_full_analysis_text = _llm_out
print(f" ✅ LLM12维分析完成({len(_full_analysis_text)}字)", flush=True)
except Exception as _e:
print(f" ❌ LLM12维分析失败: {_e}", file=__import__('sys').stderr)
_full_analysis_text = None
# 保存到DB ⚠️ 输出纪律(必须遵守):
1. 直接以【维持或修改】开头,禁止任何寒暄、开场白、分隔线
2. 禁止输出 <structured_data> 或任何 XML/JSON/代码块
3. 所有【】节标题一个都不能少"""
_full_analysis_text = None
try:
_llm_result = call_llm(_prompt, max_tokens=4096, timeout=150, retries=1, backoff=20)
if _llm_result["ok"]:
_full_analysis_text = _llm_result["content"]
print(f" ✅ LLM12维分析完成({len(_full_analysis_text)}字, {_llm_result['elapsed']:.1f}s)", flush=True)
else:
print(f" ❌ LLM12维分析失败({_llm_result['attempts']}次): {_llm_result['error'][:200]}", flush=True)
except Exception as _e:
print(f" ❌ LLM12维分析异常: {_e}", flush=True)
# ── 保存到DB(覆写前先快照)──
_fa_conn = __import__('sqlite3').connect("/home/hmo/MoFin/data/mofin.db") _fa_conn = __import__('sqlite3').connect("/home/hmo/MoFin/data/mofin.db")
_fa_conn.execute("UPDATE holding_strategies SET full_analysis=?, reassessed_at=? WHERE code=? AND status='active'", (_full_analysis_text, __import__('datetime').datetime.now().isoformat(), code)) if _full_analysis_text:
# 快照旧策略(使用共享函数)
try:
snapshot_strategy_history(_fa_conn, code, "per_stock_12d")
except Exception as _se:
print(f" ⚠️ 快照失败: {_se}", flush=True)
_fa_conn.execute(
"UPDATE holding_strategies SET full_analysis=?, reassessed_at=? WHERE code=? AND status='active'",
(_full_analysis_text, __import__('datetime').datetime.now().isoformat(), code))
_fa_conn.commit() _fa_conn.commit()
_fa_conn.close() _fa_conn.close()
print(f" ✅ 完整12维分析已保存({len(_full_analysis_text)}字)" if _full_analysis_text else f" ⚠️ 12维分析未完成,跳过保存") if _full_analysis_text:
print(f" ✅ 完整12维分析已保存({len(_full_analysis_text)}字)")
else:
print(f" ⚠️ 12维分析未完成,跳过保存")
print(f" [DB] holding_strategies 已更新: {code}") print(f" [DB] holding_strategies 已更新: {code}")
# 从LLM输出提取信号 # 从LLM输出提取信号
if _full_analysis_text and '【综合结论】' in _full_analysis_text: if _full_analysis_text and '【综合结论】' in _full_analysis_text:
@@ -490,10 +567,8 @@ def main():
"SELECT name, price, entry_low, entry_high, stop_loss, take_profit, position_advice FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone() "SELECT name, price, entry_low, entry_high, stop_loss, take_profit, position_advice FROM holding_strategies WHERE code=? AND status='active'", (code,)).fetchone()
if _nr2: if _nr2:
_xm = f"📈 {_nr2[0] or code}({code}) 价{_nr2[1]}→12维买入信号!区间{_nr2[2]}~{_nr2[3]}{_nr2[4]}{_nr2[5]} 仓位{_nr2[6] or '-'}" _xm = f"📈 {_nr2[0] or code}({code}) 价{_nr2[1]}→12维买入信号!区间{_nr2[2]}~{_nr2[3]}{_nr2[4]}{_nr2[5]} 仓位{_nr2[6] or '-'}"
_xr = __import__('urllib.request').Request("http://127.0.0.1:5805/", from alert_helper import notify as _notify2, ACTION as _ACT2
data=__import__('json').dumps({"body": _xm, "to": "hmo@yoin.fun", "type": "chat"}).encode(), _notify2("买入信号", _xm, _ACT2)
headers={"Content-Type": "application/json"})
__import__('urllib.request').urlopen(_xr, timeout=5)
print(f" 📨 XMPP推送买入信号") print(f" 📨 XMPP推送买入信号")
except: pass except: pass
except: pass except: pass
+11 -3
View File
@@ -142,7 +142,7 @@ def main():
reassess_scripts.append(code) reassess_scripts.append(code)
print(f"[AUTO_REASSESS] {name}({code}) 价{cur_price:.2f}偏离买入区中心{center:.2f} {drift:+.0f}% → 触发重评") print(f"[AUTO_REASSESS] {name}({code}) 价{cur_price:.2f}偏离买入区中心{center:.2f} {drift:+.0f}% → 触发重评")
if reassess_scripts: if reassess_scripts:
# 调用 per_stock_reassess # 调用 per_stock_reassess(每轮最多5只,防LLM慢导致整批超时;其余下轮继续)
reassess_path = None reassess_path = None
for p in ['/home/hmo/MoFin/scripts/per_stock_reassess.py', for p in ['/home/hmo/MoFin/scripts/per_stock_reassess.py',
'/home/hmo/.hermes/profiles/position-analyst/scripts/per_stock_reassess.py']: '/home/hmo/.hermes/profiles/position-analyst/scripts/per_stock_reassess.py']:
@@ -150,12 +150,20 @@ def main():
reassess_path = p reassess_path = p
break break
if reassess_path: if reassess_path:
for code in reassess_scripts: MAX_PER_RUN = 5
batch = reassess_scripts[:MAX_PER_RUN]
if len(reassess_scripts) > MAX_PER_RUN:
print(f"[AUTO_REASSESS] 本轮限{MAX_PER_RUN}只,剩余{len(reassess_scripts)-MAX_PER_RUN}只下轮继续")
for code in batch:
try:
# LLM 重评冷启动 20-100sdeepseek-v4-pro 更慢 → 480s
r = subprocess.run(['python3', reassess_path, code], r = subprocess.run(['python3', reassess_path, code],
capture_output=True, text=True, timeout=60) capture_output=True, text=True, timeout=480)
out = r.stdout.strip()[:200] if r.stdout else "" out = r.stdout.strip()[:200] if r.stdout else ""
err = r.stderr.strip()[:200] if r.stderr else "" err = r.stderr.strip()[:200] if r.stderr else ""
print(f"{code}: exited={r.returncode} {out}") print(f"{code}: exited={r.returncode} {out}")
except subprocess.TimeoutExpired:
print(f"{code}: 超时480s(LLM仍慢),下轮重试")
except Exception as e: except Exception as e:
print(f"[AUTO_REASSESS FAIL] {e}") print(f"[AUTO_REASSESS FAIL] {e}")
# ----- 结束 自选股重评 ----- # ----- 结束 自选股重评 -----