知微
80d59c9331
- deploy/bot/ — XMPP bot核心(xmpp_agent_core + xmpp_zhiwei_bot) - deploy/profile-scripts/ — cron脚本(price_monitor等) - 运行时文件已替换为指向MoFin的符号链接 - 改代码只需改MoFin，系统自动生效
273 lines
12 KiB
Python
273 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""mofin_collect.py — MoFin 数据采集链
|
|
|
|
每轮盯盘 cron 前运行，顺序执行：
|
|
0. 策略新鲜度强制检查（代码级约束：推荐前必须先检查策略是否最新）
|
|
1. market_watch — 拉90个行业板块数据（9:30前跳过，市场未开）
|
|
2. trend_detector — 检测17种信号（依赖板块数据，同跳）
|
|
3. mofin_news — 搜新闻+小果分析
|
|
4. stock_quote — 所有持仓最新行情（CRITICAL: LLM唯一价格源）
|
|
"""
|
|
|
|
import subprocess, sys, time, json
|
|
from pathlib import Path
|
|
from datetime import datetime
|
|
from mo_data import get_price, get_prices_batch
|
|
|
|
BASE = Path(__file__).parent.parent if "hermes" in str(Path(__file__).resolve()) else Path(__file__).parent
|
|
|
|
now = datetime.now()
|
|
market_open = (now.hour >= 9 and now.minute >= 30) or now.hour >= 10
|
|
|
|
# ── 步骤0: 策略新鲜度强制检查（代码级约束）──
|
|
# 在LLM看到任何数据前，先确保所有持仓策略是新鲜的
|
|
# 策略过期(>4h未更新) → 创建TODO触发重评 + 注入STALE标记到LLM context
|
|
# LLM看到STALE标记：禁止对该股给出任何操作建议
|
|
print("--- strategy_freshness_check ---", flush=True)
|
|
try:
|
|
sys.path.insert(0, str(BASE))
|
|
from mofin_db import get_conn
|
|
conn = get_conn()
|
|
cur = conn.cursor()
|
|
# 读所有活跃持仓及其最新策略更新时间
|
|
rows = cur.execute("""
|
|
SELECT h.code, h.name, h.price, h.cost, h.shares, h.position_pct,
|
|
hs.stop_loss, hs.take_profit, hs.entry_low, hs.entry_high,
|
|
hs.created_at, hs.action
|
|
FROM holdings h
|
|
LEFT JOIN (
|
|
SELECT code, stop_loss, take_profit, entry_low, entry_high,
|
|
created_at, action,
|
|
ROW_NUMBER() OVER (PARTITION BY code ORDER BY id DESC) AS rn
|
|
FROM holding_strategies
|
|
) hs ON h.code = hs.code AND hs.rn = 1
|
|
WHERE h.is_active = 1
|
|
ORDER BY h.code
|
|
""").fetchall()
|
|
|
|
stale_count = 0
|
|
fresh_count = 0
|
|
no_strategy_count = 0
|
|
|
|
for r in rows:
|
|
code = r["code"]
|
|
name = r["name"]
|
|
last_update = r["created_at"]
|
|
has_strategy = last_update is not None
|
|
|
|
if has_strategy:
|
|
try:
|
|
last_dt = datetime.fromisoformat(last_update)
|
|
hours_stale = (now - last_dt).total_seconds() / 3600
|
|
except:
|
|
hours_stale = 999
|
|
|
|
if hours_stale > 4:
|
|
stale_count += 1
|
|
# 强制代码级约束：立即执行重评，不等TODO异步处理
|
|
# 在LLM看到数据前，策略必须是新鲜的
|
|
try:
|
|
from strategy_lifecycle import reassess_with_context
|
|
result = reassess_with_context(
|
|
code, name, r["price"],
|
|
r["cost"] if r["cost"] else 0, r["shares"] if r["shares"] else 0,
|
|
r["action"] or ""
|
|
)
|
|
if result and result.get("action"):
|
|
print(f" 🔄 FORCE_REASSESS {code} {name}: {hours_stale:.0f}h过期→已立即重评→{result['action'][:60]}", flush=True)
|
|
# 写strategy_evaluations（独立短连接，防锁冲突）
|
|
try:
|
|
eval_conn = sqlite3.connect(str(Path(__file__).parent.parent / "data" / "mofin.db"), timeout=10)
|
|
eval_conn.execute("""
|
|
INSERT INTO strategy_evaluations
|
|
(code, eval_type, status, new_stop_loss, new_tp, reason, created_at)
|
|
VALUES (?, 'reassess', 'completed', ?, ?, ?, ?)
|
|
""", (
|
|
code,
|
|
result.get("stop_loss"),
|
|
result.get("take_profit"),
|
|
f"{result.get('action','')} RR={result.get('rr_ratio','?')} 信号={result.get('timing_signal','')}",
|
|
now.isoformat()
|
|
))
|
|
eval_conn.commit()
|
|
eval_conn.close()
|
|
except Exception:
|
|
pass
|
|
else:
|
|
print(f" ⚠️ FORCE_REASSESS {code} {name}: 重评返回空结果", flush=True)
|
|
except Exception as e:
|
|
print(f" ❌ FORCE_REASSESS {code} {name} 失败: {e}", flush=True)
|
|
# 创建TODO作为兜底
|
|
todo_sql = """
|
|
INSERT OR IGNORE INTO todos
|
|
(title, code, fix_action, source, priority, status, created_at)
|
|
VALUES (?, ?, 'reassess_strategy', 'freshness_check', 'high', 'pending', ?)
|
|
"""
|
|
cur.execute(todo_sql, (
|
|
f"策略过期需重评: {code} {name} ({hours_stale:.0f}h未更新)",
|
|
code,
|
|
now.isoformat()
|
|
))
|
|
print(f" ⚠️ STALE {code} {name}: {hours_stale:.0f}h未更新 → 已强制重评", flush=True)
|
|
else:
|
|
fresh_count += 1
|
|
print(f" ✅ FRESH {code} {name}: {hours_stale:.1f}h前更新", flush=True)
|
|
else:
|
|
no_strategy_count += 1
|
|
# 无策略→立即执行重评创建策略（代码级约束：不允许无策略就输出建议）
|
|
try:
|
|
from strategy_lifecycle import reassess_with_context
|
|
result = reassess_with_context(
|
|
code, name, r["price"],
|
|
r["cost"] if r["cost"] else 0, r["shares"] if r["shares"] else 0,
|
|
""
|
|
)
|
|
if result and result.get("action"):
|
|
print(f" 🔄 CREATE_STRATEGY {code} {name}: 无策略→已创建→{result['action'][:60]}", flush=True)
|
|
try:
|
|
eval_conn = sqlite3.connect(str(Path(__file__).parent.parent / "data" / "mofin.db"), timeout=10)
|
|
eval_conn.execute("""
|
|
INSERT INTO strategy_evaluations
|
|
(code, eval_type, status, new_stop_loss, new_tp, reason, created_at)
|
|
VALUES (?, 'reassess', 'completed', ?, ?, ?, ?)
|
|
""", (
|
|
code,
|
|
result.get("stop_loss"),
|
|
result.get("take_profit"),
|
|
f"{result.get('action','')} RR={result.get('rr_ratio','?')} 信号={result.get('timing_signal','')}",
|
|
now.isoformat()
|
|
))
|
|
eval_conn.commit()
|
|
eval_conn.close()
|
|
except Exception:
|
|
pass
|
|
else:
|
|
print(f" ⚠️ CREATE_STRATEGY {code} {name}: 重评返回空", flush=True)
|
|
except Exception as e:
|
|
print(f" ❌ CREATE_STRATEGY {code} {name} 失败: {e}", flush=True)
|
|
|
|
conn.commit()
|
|
|
|
# === 自选股策略检查 + 强制重评 ===
|
|
wl_fresh = 0
|
|
wl_stale = 0
|
|
wl_error = 0
|
|
try:
|
|
for wr in conn.execute("SELECT code, name, price, entry_low, entry_high, stop_loss FROM watchlist_stocks WHERE is_active=1"):
|
|
code = wr["code"]
|
|
name = wr["name"]
|
|
wl_price = wr["price"] or 0
|
|
# 自选股price可能为0（新加入未更新），从实时API获取
|
|
if wl_price <= 0:
|
|
try:
|
|
p, _ = get_price(code)
|
|
if p and p > 0:
|
|
wl_price = p
|
|
except Exception:
|
|
pass
|
|
# 自选股无cost/shares，传0
|
|
try:
|
|
from strategy_lifecycle import reassess_with_context
|
|
result = reassess_with_context(
|
|
code, name, wl_price,
|
|
0, 0, ""
|
|
)
|
|
if result and result.get("action"):
|
|
wl_stale += 1
|
|
print(f" 📋 WATCHLIST_REASSESS {code} {name}: →{result['action'][:60]}", flush=True)
|
|
try:
|
|
eval_conn = sqlite3.connect(str(Path(__file__).parent.parent / "data" / "mofin.db"), timeout=10)
|
|
eval_conn.execute("""
|
|
INSERT INTO strategy_evaluations
|
|
(code, eval_type, status, new_stop_loss, new_tp, reason, created_at)
|
|
VALUES (?, 'reassess', 'completed', ?, ?, ?, ?)
|
|
""", (
|
|
code,
|
|
result.get("stop_loss"),
|
|
result.get("take_profit"),
|
|
f"自选:{result.get('action','')} RR={result.get('rr_ratio','?')} 信号={result.get('timing_signal','')}",
|
|
now.isoformat()
|
|
))
|
|
eval_conn.commit()
|
|
eval_conn.close()
|
|
except Exception:
|
|
pass
|
|
else:
|
|
wl_fresh += 1
|
|
except Exception as e:
|
|
wl_error += 1
|
|
print(f" ❌ WATCHLIST_REASSESS {code} {name} 失败: {e}", flush=True)
|
|
except Exception as e:
|
|
print(f" ⚠️ 自选股检查跳过: {e}", flush=True)
|
|
|
|
conn.close()
|
|
|
|
total = len(rows)
|
|
wl_total = wl_fresh + wl_stale + wl_error
|
|
print(f"策略检查完成: {total}只持仓({fresh_count}新鲜/{stale_count}过期/{no_strategy_count}无策略) + {wl_total}只自选({wl_fresh}无需/{wl_stale}已重评/{wl_error}失败)", flush=True)
|
|
if stale_count > 0 or no_strategy_count > 0 or wl_stale > 0:
|
|
print(f"⚠️ 重评完成: {stale_count + no_strategy_count}只已强制刷新, LLM可基于最新策略给出建议", flush=True)
|
|
except Exception as e:
|
|
print(f"WARN: strategy_freshness_check跳过 ({e})", flush=True)
|
|
|
|
# 步骤1-3: 行业/新闻数据
|
|
SCRIPTS = []
|
|
if market_open:
|
|
SCRIPTS.append(("market_watch.py", 60))
|
|
SCRIPTS.append(("trend_detector.py", 60))
|
|
else:
|
|
print(f"[{now.strftime('%H:%M')}] 市场未开盘(9:30)，跳过板块采集", flush=True)
|
|
|
|
SCRIPTS.append(("mofin_news.py", 50))
|
|
|
|
for script, timeout in SCRIPTS:
|
|
path = BASE / script
|
|
if not path.exists():
|
|
path = Path("/home/hmo/MoFin") / script
|
|
print(f"--- {script} ---", flush=True)
|
|
start = time.time()
|
|
try:
|
|
result = subprocess.run(
|
|
[sys.executable, str(path)],
|
|
capture_output=True, text=True, timeout=timeout
|
|
)
|
|
elapsed = time.time() - start
|
|
if result.returncode == 0:
|
|
print(f"OK ({elapsed:.0f}s)", flush=True)
|
|
if result.stdout.strip():
|
|
for line in result.stdout.strip().split("\n")[-3:]:
|
|
print(f" {line}", flush=True)
|
|
else:
|
|
print(f"FAIL ({elapsed:.0f}s): {result.stderr[:200]}", flush=True)
|
|
except subprocess.TimeoutExpired:
|
|
print(f"TIMEOUT ({timeout}s)", flush=True)
|
|
except Exception as e:
|
|
print(f"ERROR: {e}", flush=True)
|
|
|
|
# ── 步骤4: 个股行情注入（唯一权威价格源）──
|
|
# 所有持仓最新行情，注入到 LLM context
|
|
# LLM 禁止自行调用原始API解析价格
|
|
PRICE_SCRIPT = BASE / "stock_quote.py"
|
|
if not PRICE_SCRIPT.exists():
|
|
PRICE_SCRIPT = Path("/home/hmo/MoFin/scripts/stock_quote.py")
|
|
if PRICE_SCRIPT.exists():
|
|
print("--- stock_quote.py ---", flush=True)
|
|
try:
|
|
result = subprocess.run(
|
|
[sys.executable, str(PRICE_SCRIPT), "--all-holdings"],
|
|
capture_output=True, text=True, timeout=30
|
|
)
|
|
if result.returncode == 0 and result.stdout.strip():
|
|
lines = [l for l in result.stdout.strip().split("\n") if l.strip()]
|
|
print(f"OK ({len(lines)}只持仓)", flush=True)
|
|
for line in lines[:50]:
|
|
print(f" {line}", flush=True)
|
|
else:
|
|
print(f"WARN: stock_quote stderr={result.stderr[:100]}", flush=True)
|
|
except Exception as e:
|
|
print(f"WARN: stock_quote skipped ({e})", flush=True)
|
|
else:
|
|
print("WARN: stock_quote.py not found", flush=True)
|
|
|
|
print("采集链完成", flush=True)
|