chore: deployed cleanup + hygiene system
This commit is contained in:
知微
@@ -16,7 +16,7 @@ try:
|
||||
except ImportError:
|
||||
HAS_AKSHARE = False
|
||||
|
||||
DATA_DIR = Path(__file__).parent.parent / "data"
|
||||
DATA_DIR = Path(__file__).parent / "data"
|
||||
DB_PATH = DATA_DIR / "mofin.db"
|
||||
MAX_ARTICLES = 5
|
||||
|
||||
|
||||
@@ -18,7 +18,6 @@ from datetime import datetime, date, timedelta
|
||||
from typing import Optional
|
||||
|
||||
from mofin_db import get_conn
|
||||
from mo_data import get_price
|
||||
|
||||
DATA_DIR = "/home/hmo/web-dashboard/data"
|
||||
HISTORY_PATH = os.path.join(DATA_DIR, "price_history.json")
|
||||
@@ -27,6 +26,9 @@ HISTORY_PATH = os.path.join(DATA_DIR, "price_history.json")
|
||||
# 腾讯API K线端点
|
||||
KLINE_URL = "http://web.ifzq.gtimg.cn/appstock/app/fqkline/get?param={market}{code},{period},,,{count},qfq"
|
||||
|
||||
# 腾讯实时行情端点（用于市场前缀判断）
|
||||
QUOTE_URL = "http://qt.gtimg.cn/q={market}{code}"
|
||||
|
||||
|
||||
def _write_klines_to_db(code: str, daily: list, weekly: list, monthly: list, fundamentals: dict = None):
|
||||
"""K线数据双写 SQLite（失败不影响缓存写入）"""
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -72,9 +72,8 @@ def detect_scenario():
|
||||
|
||||
try:
|
||||
# 优先 DB
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
db = sqlite3.connect(str(Path(__file__).parent.parent / "data" / "mofin.db"))
|
||||
from mofin_db import get_conn
|
||||
db = get_conn()
|
||||
mrow = db.execute(
|
||||
"SELECT indices, structure, sector_mood FROM macro_context_log "
|
||||
"WHERE has_valid_data=1 ORDER BY created_at DESC LIMIT 1"
|
||||
|
||||
@@ -0,0 +1,266 @@
|
||||
#!/usr/bin/env python3
|
||||
"""system_hygiene_audit.py — 系统卫生审计（防冗余复发）
|
||||
|
||||
每周运行。检查六类问题，输出 hygiene_report.json，有问题时推 XMPP。
|
||||
红线6/7/8/9/10 的自动化 enforcement。
|
||||
|
||||
检查项：
|
||||
1. 分叉副本：同名 .py 在不同权威位置内容不一致
|
||||
2. 断裂硬链接：deploy/profile-scripts vs profile scripts 内容不一致（cron 会跑旧代码！）
|
||||
3. 僵尸进程：>3 天的 python/node 进程不在白名单
|
||||
4. 孤儿数据文件：生产数据目录 >14 天未修改且不在活文件注册表
|
||||
5. 死 cron：jobs.json 中 script 不存在
|
||||
6. DB 表新鲜度：核心表 >24h 无新记录（交易时间）
|
||||
"""
|
||||
import os, sys, json, glob, hashlib, sqlite3, subprocess
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, '/home/hmo/MoFin')
|
||||
|
||||
DEPLOY = '/home/hmo/MoFin/deploy/profile-scripts'
|
||||
PA_SCRIPTS = '/home/hmo/.hermes/profiles/position-analyst/scripts'
|
||||
MOFIN_ROOT = '/home/hmo/MoFin'
|
||||
DATA_DIR = '/home/hmo/MoFin/data'
|
||||
REPORT = '/home/hmo/MoFin/gateway/logs/hygiene_report.json'
|
||||
|
||||
# 活文件注册表（生产数据目录允许存在的非数据文件）
|
||||
LIVE_DATA_FILES = {
|
||||
'mofin.db', 'mofin.db-shm', 'mofin.db-wal', 'mofin_health.json', 'portfolio.json',
|
||||
'preflight_result.json', 'growth_registry.json', 'hardcode_audit.json',
|
||||
'health_checklist.json', 'macro_divergence_state.json', 'macro_risk_state.json',
|
||||
'market.json', 'scanner_state.json', 'state.db', 'strategy_staleness_report.json',
|
||||
'system_audit_report.json', 'price_history.json', 'evaluation.json',
|
||||
'accuracy_stats.json', 'format_error_library.json', 'candidate_pool.json',
|
||||
'pipeline_registry.json', 'analyst-knowledge-log.md', 'mofin_health.html',
|
||||
'evaluation_input.json', 'push_cooldown.json', 'system_audit.json',
|
||||
'system_inventory.json', 'stocks',
|
||||
}
|
||||
|
||||
PROCESS_WHITELIST = [
|
||||
'hermes_cli.main', 'server.py', 'xmpp_zhiwei_bot.py', 'xmpp_mohe_bot.py',
|
||||
'shadowsocks', 'unattended-upgrades', 'dashboard.py', 'kanban_api.py',
|
||||
'main_dsa.py', 'vc-webhook.py', 'obsidian-api.py', 'http.server',
|
||||
'wechat_webhook.py', 'qq-poll', 'afw', 'mcp_server', 'uvicorn',
|
||||
'agentmemory', 'kimi_collect', 'mohe_knowledge_relay', 'wechat_watchdog',
|
||||
'todo_scanner', 'miner.py',
|
||||
]
|
||||
|
||||
CORE_TABLES = [
|
||||
('live_prices', 'updated_at', '实时价格'),
|
||||
('market_snapshots', 'created_at', '市场快照'),
|
||||
('mtf_cache', 'updated_at', '多周期缓存'),
|
||||
('macro_context_log', 'created_at', '宏观上下文'),
|
||||
('price_events', 'created_at', '价格事件'),
|
||||
]
|
||||
|
||||
|
||||
def md5(p):
|
||||
try:
|
||||
return hashlib.md5(open(p, 'rb').read()).hexdigest()
|
||||
except Exception:
|
||||
return 'ERR'
|
||||
|
||||
|
||||
def check_diverged():
|
||||
"""检查 deploy vs MoFin/scripts vs MoFin根 的分叉副本"""
|
||||
issues = []
|
||||
compare_dirs = [f'{MOFIN_ROOT}/scripts', MOFIN_ROOT, '/home/hmo/web-dashboard']
|
||||
for f in os.listdir(DEPLOY):
|
||||
if not f.endswith('.py'):
|
||||
continue
|
||||
dp = os.path.join(DEPLOY, f)
|
||||
d_md5 = md5(dp)
|
||||
for d in compare_dirs:
|
||||
p = os.path.join(d, f)
|
||||
if os.path.exists(p) and not os.path.islink(p):
|
||||
try:
|
||||
if os.path.samefile(p, dp):
|
||||
continue
|
||||
except Exception:
|
||||
pass
|
||||
if md5(p) != d_md5:
|
||||
issues.append({
|
||||
'type': 'diverged_copy', 'file': f,
|
||||
'canonical': dp, 'stale_copy': p,
|
||||
'action': f'归档 {p} 或硬链接到权威版',
|
||||
})
|
||||
return issues
|
||||
|
||||
|
||||
def check_broken_hardlinks():
|
||||
"""deploy vs pa/scripts 内容不一致 = cron 跑旧代码"""
|
||||
issues = []
|
||||
for f in os.listdir(DEPLOY):
|
||||
if not f.endswith('.py'):
|
||||
continue
|
||||
dp = os.path.join(DEPLOY, f)
|
||||
pp = os.path.join(PA_SCRIPTS, f)
|
||||
if os.path.exists(pp):
|
||||
try:
|
||||
if os.path.samefile(dp, pp):
|
||||
continue
|
||||
except Exception:
|
||||
pass
|
||||
if md5(dp) != md5(pp):
|
||||
issues.append({
|
||||
'type': 'broken_hardlink', 'file': f,
|
||||
'action': 'bash deploy/profile-scripts/sync_profile_scripts.sh',
|
||||
})
|
||||
else:
|
||||
issues.append({
|
||||
'type': 'missing_profile_link', 'file': f,
|
||||
'action': 'bash deploy/profile-scripts/sync_profile_scripts.sh',
|
||||
})
|
||||
return issues
|
||||
|
||||
|
||||
def check_zombies():
|
||||
""">3 天的 python/node 进程不在白名单（docker 容器内进程豁免）"""
|
||||
issues = []
|
||||
try:
|
||||
r = subprocess.run(['ps', '-eo', 'pid,etime,args'], capture_output=True, text=True, timeout=10)
|
||||
for line in r.stdout.splitlines()[1:]:
|
||||
parts = line.split(None, 2)
|
||||
if len(parts) < 3:
|
||||
continue
|
||||
pid, etime, cmd = parts
|
||||
if 'python' not in cmd and 'node' not in cmd:
|
||||
continue
|
||||
# docker 容器内进程豁免（cgroup 含 docker）
|
||||
try:
|
||||
cg = open(f'/proc/{pid}/cgroup').read()
|
||||
if 'docker' in cg:
|
||||
continue
|
||||
except Exception:
|
||||
pass
|
||||
# etime 格式: dd-hh:mm:ss 或 hh:mm:ss
|
||||
days = 0
|
||||
if '-' in etime:
|
||||
days = int(etime.split('-')[0])
|
||||
if days >= 3:
|
||||
if not any(w in cmd for w in PROCESS_WHITELIST):
|
||||
issues.append({
|
||||
'type': 'zombie_process', 'pid': pid, 'days': days,
|
||||
'cmd': cmd[:120],
|
||||
'action': f'确认后 kill {pid}（红线9 收尸流程）',
|
||||
})
|
||||
except Exception:
|
||||
pass
|
||||
return issues
|
||||
|
||||
|
||||
def check_orphan_files():
|
||||
"""生产数据目录的孤儿文件"""
|
||||
issues = []
|
||||
now = datetime.now()
|
||||
for f in os.listdir(DATA_DIR):
|
||||
p = os.path.join(DATA_DIR, f)
|
||||
if not os.path.isfile(p) or f.startswith('.'):
|
||||
continue
|
||||
if f in LIVE_DATA_FILES:
|
||||
continue
|
||||
age_d = (now.timestamp() - os.path.getmtime(p)) / 86400
|
||||
if age_d > 14:
|
||||
issues.append({
|
||||
'type': 'orphan_data_file', 'file': f,
|
||||
'age_days': round(age_d),
|
||||
'action': f'归档到 archive/（红线8）',
|
||||
})
|
||||
return issues
|
||||
|
||||
|
||||
def check_dead_cron():
|
||||
"""cron job 指向不存在的脚本"""
|
||||
issues = []
|
||||
for jf, sdir in [('/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json', PA_SCRIPTS),
|
||||
('/home/hmo/.hermes/cron/jobs.json', '/home/hmo/.hermes/scripts')]:
|
||||
try:
|
||||
d = json.load(open(jf))
|
||||
jobs = d if isinstance(d, list) else d.get('jobs', [])
|
||||
for j in jobs:
|
||||
s = j.get('script')
|
||||
if s and j.get('enabled', True) and not os.path.exists(os.path.join(sdir, s)):
|
||||
issues.append({
|
||||
'type': 'dead_cron_script', 'job': j.get('name'), 'script': s,
|
||||
'action': '删除 job 或补齐脚本',
|
||||
})
|
||||
except Exception:
|
||||
pass
|
||||
return issues
|
||||
|
||||
|
||||
def check_db_freshness():
|
||||
"""核心表新鲜度（红线10）"""
|
||||
issues = []
|
||||
try:
|
||||
c = sqlite3.connect(os.path.join(DATA_DIR, 'mofin.db'), timeout=10)
|
||||
now = datetime.now()
|
||||
is_trading_time = now.weekday() < 5 and 9 <= now.hour <= 16
|
||||
for table, col, label in CORE_TABLES:
|
||||
try:
|
||||
row = c.execute(f"SELECT MAX({col}) FROM {table}").fetchone()
|
||||
if row and row[0]:
|
||||
last = datetime.fromisoformat(str(row[0]).replace('Z', ''))
|
||||
age_h = (now - last).total_seconds() / 3600
|
||||
threshold = 4 if is_trading_time else 48
|
||||
if age_h > threshold:
|
||||
issues.append({
|
||||
'type': 'stale_table', 'table': table, 'label': label,
|
||||
'age_hours': round(age_h, 1), 'threshold': threshold,
|
||||
'action': '查对应采集脚本的 cron 状态',
|
||||
})
|
||||
else:
|
||||
issues.append({'type': 'empty_table', 'table': table, 'label': label,
|
||||
'action': '查采集链路'})
|
||||
except Exception:
|
||||
pass
|
||||
c.close()
|
||||
except Exception as e:
|
||||
issues.append({'type': 'db_error', 'error': str(e)[:100]})
|
||||
return issues
|
||||
|
||||
|
||||
def main():
|
||||
print('🧹 系统卫生审计', datetime.now().strftime('%Y-%m-%d %H:%M'))
|
||||
all_issues = []
|
||||
for name, fn in [('分叉副本', check_diverged), ('断裂硬链接', check_broken_hardlinks),
|
||||
('僵尸进程', check_zombies), ('孤儿文件', check_orphan_files),
|
||||
('死cron', check_dead_cron), ('DB新鲜度', check_db_freshness)]:
|
||||
found = fn()
|
||||
status = f'❌ {len(found)}' if found else '✅'
|
||||
print(f' {status} {name}')
|
||||
all_issues.extend(found)
|
||||
|
||||
report = {
|
||||
'generated_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
||||
'issue_count': len(all_issues),
|
||||
'issues': all_issues,
|
||||
'status': 'warn' if all_issues else 'ok',
|
||||
}
|
||||
os.makedirs(os.path.dirname(REPORT), exist_ok=True)
|
||||
with open(REPORT, 'w', encoding='utf-8') as f:
|
||||
json.dump(report, f, ensure_ascii=False, indent=2)
|
||||
|
||||
if all_issues:
|
||||
# 推 XMPP
|
||||
try:
|
||||
import urllib.request
|
||||
lines = [f"🧹 系统卫生审计发现 {len(all_issues)} 个问题："]
|
||||
for i in all_issues[:8]:
|
||||
lines.append(f"• [{i['type']}] {i.get('file') or i.get('job') or i.get('table') or i.get('pid')}: {i.get('action','')[:60]}")
|
||||
if len(all_issues) > 8:
|
||||
lines.append(f'… 共 {len(all_issues)} 个，详见 hygiene_report.json')
|
||||
payload = json.dumps({'to': 'hmo@yoin.fun', 'body': '\n'.join(lines), 'type': 'chat'}).encode()
|
||||
req = urllib.request.Request('http://127.0.0.1:5805/', data=payload,
|
||||
headers={'Content-Type': 'application/json'})
|
||||
urllib.request.urlopen(req, timeout=5)
|
||||
print(' 📨 已推 XMPP')
|
||||
except Exception as e:
|
||||
print(f' XMPP 推送失败: {e}')
|
||||
else:
|
||||
print(' ✅ 系统卫生良好')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -12,8 +12,19 @@
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import urllib.request
|
||||
from datetime import datetime, date
|
||||
from mo_data import get_price
|
||||
|
||||
# 腾讯API字段索引
|
||||
F = {
|
||||
"name": 1, "code": 2, "price": 3, "close_yest": 4, "open": 5,
|
||||
"volume": 6, "timestamp": 30, "change": 31, "change_pct": 32,
|
||||
"high": 33, "low": 34, "amplitude": 43,
|
||||
"turnover": 38, "pe": 39, "pb": 46,
|
||||
"limit_up": 47, "limit_down": 48,
|
||||
"avg_price": 51, "inner_vol": 52, "outer_vol": 53,
|
||||
}
|
||||
|
||||
HISTORY_PATH = "/home/hmo/web-dashboard/data/price_history.json"
|
||||
HISTORY_DAYS = 60 # 使用最近 N 天的 HLC 数据
|
||||
@@ -44,7 +55,7 @@ def _market_prefix(code):
|
||||
|
||||
|
||||
def get_quote(code):
|
||||
"""获取行情数据。使用 mo_data.get_price 统一入口，缓存+格式转换"""
|
||||
"""获取行情数据。先拿DB的价格和涨跌幅，再调腾讯API拿HLC全量数据"""
|
||||
import time
|
||||
_cache = get_quote.__dict__.get("_cache", {})
|
||||
now = time.time()
|
||||
@@ -52,51 +63,88 @@ def get_quote(code):
|
||||
if cached and (now - cached["ts"]) < 60:
|
||||
return cached["data"]
|
||||
|
||||
price, change_pct = get_price(code)
|
||||
if price is None:
|
||||
return {"code": code, "error": "价格获取失败"}
|
||||
# 先从DB拿基础价格（快速，不阻塞）
|
||||
db_price = None
|
||||
db_chg = None
|
||||
try:
|
||||
from mofin_db import get_price_from_db
|
||||
p, chg = get_price_from_db(code)
|
||||
if p:
|
||||
db_price, db_chg = p, chg
|
||||
except:
|
||||
pass
|
||||
|
||||
# 腾讯API获取全量HLC数据
|
||||
raw = str(code).split("_")[0]
|
||||
prefix = _market_prefix(code)
|
||||
today_str = date.today().isoformat()
|
||||
url = f"http://qt.gtimg.cn/q={prefix}{raw}"
|
||||
try:
|
||||
r = urllib.request.urlopen(url, timeout=5)
|
||||
fields = r.read().decode("gbk").split('"')[1].split("~")
|
||||
except Exception as e:
|
||||
if db_price:
|
||||
return {"code": code, "price": db_price, "change_pct": db_chg or 0}
|
||||
return {"code": code, "error": str(e)}
|
||||
|
||||
def get(i):
|
||||
try:
|
||||
return float(fields[i]) if fields[i].strip() else None
|
||||
except (IndexError, ValueError):
|
||||
return None
|
||||
|
||||
today_str = date.today().isoformat()
|
||||
q = {
|
||||
"code": raw,
|
||||
"market": prefix,
|
||||
"name": code,
|
||||
"price": price,
|
||||
"close_yest": None,
|
||||
"open": None,
|
||||
"high": None,
|
||||
"low": None,
|
||||
"volume": None,
|
||||
"amount": None,
|
||||
"change": None,
|
||||
"change_pct": change_pct or 0,
|
||||
"amplitude": None,
|
||||
"turnover_rate": None,
|
||||
"pe": None,
|
||||
"pb": None,
|
||||
"limit_up": None,
|
||||
"limit_down": None,
|
||||
"avg_price": None,
|
||||
"inner_vol": None,
|
||||
"outer_vol": None,
|
||||
"timestamp": "",
|
||||
"name": fields[F["name"]] if len(fields) > F["name"] else code,
|
||||
"price": get(3),
|
||||
"close_yest": get(4),
|
||||
"open": get(5),
|
||||
"high": get(33),
|
||||
"low": get(34),
|
||||
"volume": get(6),
|
||||
"amount": get(37),
|
||||
"change": get(31),
|
||||
"change_pct": get(32),
|
||||
"amplitude": get(43),
|
||||
"turnover_rate": get(38),
|
||||
"pe": get(39),
|
||||
"pb": get(46),
|
||||
"limit_up": get(47),
|
||||
"limit_down": get(48),
|
||||
"avg_price": get(51),
|
||||
"inner_vol": get(52),
|
||||
"outer_vol": get(53),
|
||||
"timestamp": fields[F["timestamp"]] if len(fields) > F["timestamp"] else "",
|
||||
"_date": today_str,
|
||||
}
|
||||
|
||||
# 写入价格历史缓存（每日一次，只存价格）
|
||||
history = _load_history()
|
||||
if raw not in history:
|
||||
history[raw] = []
|
||||
days = history[raw]
|
||||
if days and len(days) > 0 and days[-1].get("date") == today_str:
|
||||
days[-1]["close"] = price
|
||||
else:
|
||||
days.append({"date": today_str, "close": price})
|
||||
history[raw] = days[-HISTORY_DAYS:]
|
||||
_save_history(history)
|
||||
# 写入价格历史缓存（每日一次）
|
||||
h = get(33) # high
|
||||
l = get(34) # low
|
||||
c = get(3) # price / close
|
||||
v = get(6) # volume（手）
|
||||
amt = get(37) # 成交额
|
||||
if h and l and c:
|
||||
history = _load_history()
|
||||
if raw not in history:
|
||||
history[raw] = []
|
||||
days = history[raw]
|
||||
# 如果今天已有记录，更新（盘中数据更精确）
|
||||
if days and len(days) > 0 and days[-1].get("date") == today_str:
|
||||
days[-1]["high"] = max(days[-1]["high"], h)
|
||||
days[-1]["low"] = min(days[-1]["low"], l)
|
||||
days[-1]["close"] = c # 盘中用最新价，收盘后是收盘价
|
||||
if v: days[-1]["volume"] = v
|
||||
if amt: days[-1]["amount"] = amt
|
||||
else:
|
||||
entry = {"date": today_str, "high": h, "low": l, "close": c}
|
||||
if v: entry["volume"] = v
|
||||
if amt: entry["amount"] = amt
|
||||
days.append(entry)
|
||||
# 只保留最近 HISTORY_DAYS 天
|
||||
history[raw] = days[-HISTORY_DAYS:]
|
||||
_save_history(history)
|
||||
|
||||
# 写入60秒缓存
|
||||
get_quote.__dict__["_cache"] = {**get_quote.__dict__.get("_cache", {}), code: {"ts": now, "data": q}}
|
||||
@@ -424,7 +472,9 @@ def analyze_volume_deep(code):
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
|
||||
DATA_DIR = Path(__file__).parent.parent / "data"
|
||||
DATA_DIR = Path(__file__).parent / "scripts" / "data"
|
||||
if not (DATA_DIR / "mofin.db").exists():
|
||||
DATA_DIR = Path(__file__).parent / "data"
|
||||
try:
|
||||
conn = sqlite3.connect(str(DATA_DIR / "mofin.db"))
|
||||
row = conn.execute("SELECT cache_json FROM mtf_cache WHERE code=?", (code,)).fetchone()
|
||||
@@ -587,30 +637,9 @@ def full_analysis(code):
|
||||
if 'weekly' in mtf_raw:
|
||||
w = mtf_raw['weekly']
|
||||
ws = w.get('support_resistance', {})
|
||||
wt = w.get('trend', {})
|
||||
wm = w.get('mas', {})
|
||||
mtf['weekly'] = {
|
||||
mtf['weekly_sr'] = {
|
||||
'weak_resist': ws.get('weak_resist'),
|
||||
'weak_support': ws.get('weak_support'),
|
||||
'strong_resist': ws.get('strong_resist'),
|
||||
'strong_support': ws.get('strong_support'),
|
||||
'trend': wt.get('direction', ''),
|
||||
'ma5': wm.get('ma5'),
|
||||
'ma10': wm.get('ma10'),
|
||||
}
|
||||
# 月线作为长期参考
|
||||
if 'monthly' in mtf_raw:
|
||||
m = mtf_raw['monthly']
|
||||
ms = m.get('support_resistance', {})
|
||||
mt = m.get('trend', {})
|
||||
mm = m.get('mas', {})
|
||||
mtf['monthly'] = {
|
||||
'weak_resist': ms.get('weak_resist'),
|
||||
'weak_support': ms.get('weak_support'),
|
||||
'strong_resist': ms.get('strong_resist'),
|
||||
'strong_support': ms.get('strong_support'),
|
||||
'trend': mt.get('direction', ''),
|
||||
'ma5': mm.get('ma5'),
|
||||
}
|
||||
except Exception:
|
||||
pass # non-critical, graceful degradation
|
||||
|
||||
@@ -1,266 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""xiaoguo_news_processor.py — 小果新闻情报处理
|
||||
|
||||
配合 trend_detector（每30分）运行，处理未处理的 sector_signals。
|
||||
|
||||
流程：
|
||||
1. 读未 processed 的 signals（每次1条）
|
||||
2. akshare 搜新闻（板块相关个股 + 持仓 + 自选）
|
||||
3. 调小果 LLM 逐批分析（每批3-5篇，给摘要+情感）
|
||||
4. 写入 signal_news
|
||||
5. 标记 signal.processed = true
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import urllib.request
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
try:
|
||||
import akshare as ak
|
||||
HAS_AKSHARE = True
|
||||
except ImportError:
|
||||
HAS_AKSHARE = False
|
||||
|
||||
DATA_DIR = Path(__file__).parent.parent / "data"
|
||||
DB_PATH = DATA_DIR / "mofin.db"
|
||||
XIAOGUO_API = "http://node122:18003/v1/chat/completions" # fallback, /etc/hosts resolves to LAN or EasyTier
|
||||
|
||||
def _get_xiaoguo_url():
|
||||
try:
|
||||
from mo_config import get_config
|
||||
return get_config().xiaoguo_api_url
|
||||
except Exception:
|
||||
return XIAOGUO_API
|
||||
XIAOGUO_MODEL = "Qwen3.6-27B-MTPLX-Optimized-Speed"
|
||||
MAX_ARTICLES = 5 # 每次最多分析篇数（实测5篇12s）
|
||||
|
||||
|
||||
def clean_proxy():
|
||||
for k in ['http_proxy', 'https_proxy', 'HTTP_PROXY', 'HTTPS_PROXY']:
|
||||
os.environ.pop(k, None)
|
||||
|
||||
|
||||
def get_conn():
|
||||
import sqlite3
|
||||
conn = sqlite3.connect(str(DB_PATH))
|
||||
conn.row_factory = sqlite3.Row
|
||||
return conn
|
||||
|
||||
|
||||
def search_akshare_news(code, max_results=3):
|
||||
"""用 akshare 搜个股新闻（含全文）"""
|
||||
articles = []
|
||||
if not HAS_AKSHARE:
|
||||
return articles
|
||||
try:
|
||||
clean_proxy()
|
||||
df = ak.stock_news_em(symbol=code)
|
||||
for _, r in df.head(max_results).iterrows():
|
||||
title = r.get('新闻标题', '')
|
||||
content = r.get('新闻内容', '')
|
||||
if title and len(title) > 5:
|
||||
articles.append({
|
||||
"title": title,
|
||||
"content": content,
|
||||
"url": r.get('新闻链接', '')
|
||||
})
|
||||
except:
|
||||
pass
|
||||
return articles
|
||||
|
||||
|
||||
def extract_json(text):
|
||||
"""从回复中提取JSON数组或对象"""
|
||||
# 先找 ```json ... ``` 代码块
|
||||
m = re.search(r'```(?:json)?\s*(\[[\s\S]*?\]|\{[\s\S]*?\})\s*```', text)
|
||||
if m:
|
||||
try:
|
||||
return json.loads(m.group(1))
|
||||
except:
|
||||
pass
|
||||
# 找第一个 [ 或 { 到最后一个 ] 或 }
|
||||
for start_ch, end_ch in [('[', ']'), ('{', '}')]:
|
||||
s = text.find(start_ch)
|
||||
if s >= 0:
|
||||
depth = 0
|
||||
for i in range(s, len(text)):
|
||||
if text[i] == start_ch:
|
||||
depth += 1
|
||||
elif text[i] == end_ch:
|
||||
depth -= 1
|
||||
if depth == 0:
|
||||
try:
|
||||
return json.loads(text[s:i+1])
|
||||
except:
|
||||
break
|
||||
return None
|
||||
|
||||
|
||||
def call_xiaoguo(articles):
|
||||
"""调小果LLM：给摘要+情感"""
|
||||
lines = []
|
||||
for a in articles:
|
||||
title = re.sub(r'\b\d{6}\b', '', a['title']).strip()
|
||||
title = re.sub(r'\s+', ' ', title)
|
||||
content = a.get('content') or ''
|
||||
# 给正文加标点分隔（akshare正文无标点，模型推理会卡）
|
||||
if content and not any(c in content for c in '。，！？；'):
|
||||
content = '。'.join([content[i:i+20] for i in range(0, len(content), 20)])
|
||||
if content:
|
||||
lines.append(f"{len(lines)+1}. {title}\n {content}")
|
||||
else:
|
||||
lines.append(f"{len(lines)+1}. {title}")
|
||||
prompt = "\n".join(lines) + "\n\n逐篇分析：给摘要（概括核心内容）和情感（positive/negative/neutral）。JSON数组。"
|
||||
|
||||
payload = json.dumps({
|
||||
"model": XIAOGUO_MODEL,
|
||||
"messages": [{"role": "user", "content": prompt}],
|
||||
"temperature": 0.1,
|
||||
"max_tokens": 2048,
|
||||
}).encode()
|
||||
|
||||
clean_proxy()
|
||||
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
||||
req = urllib.request.Request(
|
||||
_get_xiaoguo_url(), data=payload,
|
||||
headers={"Content-Type": "application/json"}, method="POST"
|
||||
)
|
||||
try:
|
||||
resp = opener.open(req, timeout=60)
|
||||
data = json.loads(resp.read())
|
||||
content = data["choices"][0]["message"]["content"]
|
||||
result = extract_json(content)
|
||||
if isinstance(result, list):
|
||||
return result
|
||||
except Exception as e:
|
||||
print(f" 小果调用失败: {e}", flush=True)
|
||||
return None
|
||||
|
||||
|
||||
def translate_sentiment(s):
|
||||
"""将英文情感转中文"""
|
||||
m = {"positive": "利好", "negative": "利空", "neutral": "中性"}
|
||||
return m.get(s.lower() if isinstance(s, str) else "", s)
|
||||
|
||||
|
||||
def fallback_classify(batch):
|
||||
"""关键词降级分类（小果API不可用时）"""
|
||||
positive_kw = ['突破', '增长', '利好', '加单', '订单', '放量', '新高', '获批', '量产',
|
||||
'超预期', '投产', '融资', '增持', '回购', '降息', '减税', '补贴',
|
||||
'国产替代', '自主可控', '准入']
|
||||
negative_kw = ['管制', '限制', '制裁', '利空', '减持', '抛售', '下跌', '跌停',
|
||||
'风险', '违约', '调查', '暂停', '取消', '下滑', '亏损', '裁员',
|
||||
'诉讼', '退市', '做空', '关税', '禁令']
|
||||
|
||||
for a in batch:
|
||||
text = a['title'] + (a.get('content') or '')
|
||||
pos = sum(1 for kw in positive_kw if kw in text)
|
||||
neg = sum(1 for kw in negative_kw if kw in text)
|
||||
if pos > neg:
|
||||
a['sentiment'] = '利好'
|
||||
elif neg > pos:
|
||||
a['sentiment'] = '利空'
|
||||
else:
|
||||
a['sentiment'] = '中性'
|
||||
a['summary'] = a['title'][:80]
|
||||
return batch
|
||||
|
||||
|
||||
def main():
|
||||
conn = get_conn()
|
||||
signals = conn.execute(
|
||||
"SELECT * FROM sector_signals WHERE processed = 0 ORDER BY severity DESC, id ASC LIMIT 1"
|
||||
).fetchall()
|
||||
|
||||
if not signals:
|
||||
print("无未处理的信号", flush=True)
|
||||
conn.close()
|
||||
return
|
||||
|
||||
signal = dict(signals[0])
|
||||
sector = signal["sector"]
|
||||
related = json.loads(signal["related_stocks"] or "[]")
|
||||
holdings = json.loads(signal["holdings_in_sector"] or "[]")
|
||||
watchlist = json.loads(signal["watchlist_in_sector"] or "[]")
|
||||
|
||||
print(f"处理信号: [{signal['severity']}] {signal['signal_type']} {sector}", flush=True)
|
||||
|
||||
codes = {}
|
||||
for item in related + holdings + watchlist:
|
||||
if item.get("code"):
|
||||
codes[item["code"]] = item.get("name", "")
|
||||
|
||||
members = conn.execute(
|
||||
"SELECT s.code, s.name FROM stocks s JOIN stock_sectors ss ON s.code=ss.code WHERE ss.sector_name=? LIMIT 5",
|
||||
(sector,)
|
||||
).fetchall()
|
||||
for m in members:
|
||||
if m["code"] not in codes:
|
||||
codes[m["code"]] = m["name"]
|
||||
|
||||
all_articles = []
|
||||
for code, name in codes.items():
|
||||
arts = search_akshare_news(code, 3)
|
||||
for a in arts:
|
||||
if a["title"] not in [x["title"] for x in all_articles]:
|
||||
all_articles.append(a)
|
||||
print(f" 搜 {name}({code}): {len(arts)} 篇", flush=True)
|
||||
|
||||
if not all_articles:
|
||||
print(" 未搜到新闻", flush=True)
|
||||
conn.execute("UPDATE sector_signals SET processed=1 WHERE id=?", (signal["id"],))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
return
|
||||
|
||||
# 只取前5篇，跳过含有表格数据的脏内容
|
||||
filtered = []
|
||||
for a in all_articles:
|
||||
c = a.get('content', '') or ''
|
||||
if any(kw in c for kw in ['主力资金', '资金净流入', '代码', '简称']):
|
||||
continue
|
||||
filtered.append(a)
|
||||
if len(filtered) >= MAX_ARTICLES:
|
||||
break
|
||||
batch = filtered[:MAX_ARTICLES]
|
||||
print(f" 共{len(all_articles)}篇，送小果分析{len(batch)}篇", flush=True)
|
||||
|
||||
results = call_xiaoguo(batch)
|
||||
if not results:
|
||||
print(" 小果API不可用，降级到关键词分类", flush=True)
|
||||
fallback_classify(batch)
|
||||
results = None # batch already has sentiment/summary set
|
||||
|
||||
if results and isinstance(results, list):
|
||||
# 小果LLM返回结果，按索引匹配
|
||||
for i, r in enumerate(results):
|
||||
if i < len(batch):
|
||||
batch[i]["sentiment"] = translate_sentiment(r.get("sentiment", r.get("情感", "")))
|
||||
batch[i]["summary"] = r.get("summary", r.get("摘要", ""))
|
||||
else:
|
||||
break
|
||||
|
||||
# 汇总情感
|
||||
sentiments = [a.get("sentiment", "中性") for a in batch if a.get("sentiment")]
|
||||
pos = sentiments.count("利好")
|
||||
neg = sentiments.count("利空")
|
||||
overall = "利好" if pos > neg * 1.5 else "利空" if neg > pos * 1.5 else "中性"
|
||||
summaries = [a.get("summary", "") for a in batch if a.get("summary")]
|
||||
combined = f"{sector}板块信号：{'｜'.join(summaries[:3])}。总体{overall}。"
|
||||
|
||||
searched_names = list(set(codes.values()))
|
||||
conn.execute(
|
||||
"INSERT INTO signal_news (signal_id, sector, overall_sentiment, summary, key_articles, searched_stocks) VALUES (?, ?, ?, ?, ?, ?)",
|
||||
(signal["id"], sector, overall, combined, json.dumps(batch, ensure_ascii=False), json.dumps(searched_names, ensure_ascii=False))
|
||||
)
|
||||
conn.execute("UPDATE sector_signals SET processed=1 WHERE id=?", (signal["id"],))
|
||||
conn.commit()
|
||||
|
||||
print(f" 完成: {overall} — {combined[:100]}", flush=True)
|
||||
conn.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,350 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""xiaoguo_scanner.py — 小果独立扫描线
|
||||
|
||||
每5分钟跑一轮，全市场排行榜主动发现潜在标的。
|
||||
不依赖 trend_detector 信号，独立产出到 signal_news。
|
||||
"""
|
||||
|
||||
import json, os, re, time, urllib.request
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
try:
|
||||
import akshare as ak
|
||||
HAS_AKSHARE = True
|
||||
except ImportError:
|
||||
HAS_AKSHARE = False
|
||||
|
||||
DATA_DIR = Path("/home/hmo/MoFin/data")
|
||||
DB_PATH = DATA_DIR / "mofin.db"
|
||||
XIAOGUO_API = "http://node122:18003/v1/chat/completions"
|
||||
XIAOGUO_MODEL = "Qwen3.6-27B-MTPLX-Optimized-Speed"
|
||||
SCAN_INTERVAL = 3600 # 同一只股1小时内不重复搜
|
||||
MAX_STOCKS_PER_RUN = 15
|
||||
ARTICLES_PER_STOCK = 3
|
||||
|
||||
# 同花顺看多榜（挖掘潜力股）
|
||||
BULLISH_BOARDS = [
|
||||
("创新高", "stock_rank_cxg_ths"),
|
||||
("量价齐升", "stock_rank_ljqs_ths"),
|
||||
("向上突破", "stock_rank_xstp_ths"),
|
||||
("连续上涨", "stock_rank_lxsz_ths"),
|
||||
("持续放量", "stock_rank_cxfl_ths"),
|
||||
("险资举牌", "stock_rank_xzjp_ths"),
|
||||
]
|
||||
|
||||
# 同花顺看空榜（持仓风险预警）
|
||||
BEARISH_BOARDS = [
|
||||
("创新低", "stock_rank_cxd_ths"),
|
||||
("持续缩量", "stock_rank_cxsl_ths"),
|
||||
("量价齐跌", "stock_rank_ljqd_ths"),
|
||||
("连续下跌", "stock_rank_lxxd_ths"),
|
||||
("向下突破", "stock_rank_xxtp_ths"),
|
||||
]
|
||||
|
||||
ALL_BOARDS = BULLISH_BOARDS + BEARISH_BOARDS
|
||||
BULLISH_COUNT = len(BULLISH_BOARDS)
|
||||
|
||||
# 行业领涨股扫描（不轮换，每轮都跑）
|
||||
# 从 market.json 读热门行业领涨股
|
||||
SECTOR_HOT_THRESHOLD = 2.5 # 板块涨幅>2.5%时捞它的领涨股
|
||||
|
||||
|
||||
def clean_proxy():
|
||||
for k in ['http_proxy','https_proxy','HTTP_PROXY','HTTPS_PROXY']:
|
||||
os.environ.pop(k, None)
|
||||
|
||||
|
||||
def get_conn():
|
||||
import sqlite3
|
||||
conn = sqlite3.connect(str(DB_PATH))
|
||||
conn.row_factory = sqlite3.Row
|
||||
return conn
|
||||
|
||||
|
||||
def fetch_hot_board():
|
||||
"""东方财富热榜"""
|
||||
if not HAS_AKSHARE:
|
||||
return []
|
||||
try:
|
||||
clean_proxy()
|
||||
df = ak.stock_hot_rank_em()
|
||||
if df is None or len(df) == 0:
|
||||
return []
|
||||
# 东方财富热榜列名变化较大，自动检测
|
||||
cols = list(df.columns)
|
||||
code_candidates = [c for c in cols if any(x in c for x in ['代码', 'code', 'CODE'])]
|
||||
name_candidates = [c for c in cols if any(x in c for x in ['简称', '名称', 'name', 'NAME'])]
|
||||
code_col = code_candidates[0] if code_candidates else cols[1]
|
||||
name_col = name_candidates[0] if name_candidates else cols[2]
|
||||
return [{"code": str(r[code_col]).zfill(6).strip(), "name": str(r[name_col]).strip(),
|
||||
"rank": i+1, "source": "东方财富热榜"}
|
||||
for i, (_, r) in enumerate(df.head(30).iterrows())]
|
||||
except Exception:
|
||||
pass
|
||||
return []
|
||||
|
||||
|
||||
def fetch_rotating_board():
|
||||
"""同花顺轮流榜（每轮一个），返回 (股票列表, 是否看多)"""
|
||||
if not HAS_AKSHARE:
|
||||
return [], True
|
||||
conn = get_conn()
|
||||
row = conn.execute("SELECT val FROM state_meta WHERE key='xiaoguo_board_round'").fetchone()
|
||||
round_idx = (int(row[0]) if row else 0) % len(ALL_BOARDS)
|
||||
conn.execute("INSERT OR REPLACE INTO state_meta (key, val) VALUES ('xiaoguo_board_round', ?)",
|
||||
(str((round_idx + 1) % len(ALL_BOARDS)),))
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
board_name, func_name = ALL_BOARDS[round_idx]
|
||||
is_bullish = round_idx < BULLISH_COUNT
|
||||
print(f" 同花顺榜: {board_name} {'📈看多' if is_bullish else '📉看空'}", flush=True)
|
||||
|
||||
try:
|
||||
clean_proxy()
|
||||
fn = getattr(ak, func_name)
|
||||
df = fn()
|
||||
cols = list(df.columns)
|
||||
code_col = [c for c in cols if '代码' in c][0]
|
||||
name_col = [c for c in cols if '简称' in c or '名称' in c][0]
|
||||
return [{"code": str(r[code_col]).zfill(6), "name": str(r[name_col]).strip(),
|
||||
"source": f"同花顺{board_name}"}
|
||||
for _, r in df.head(15).iterrows()], is_bullish
|
||||
except Exception as e:
|
||||
print(f" {board_name}失败: {e}", flush=True)
|
||||
return [], is_bullish
|
||||
|
||||
|
||||
def fetch_sector_leaders():
|
||||
"""从 market.json 读热门行业领涨股"""
|
||||
mkt_path = DATA_DIR / "market.json"
|
||||
if not mkt_path.exists():
|
||||
return []
|
||||
try:
|
||||
mkt = json.loads(mkt_path.read_text())
|
||||
sectors = mkt.get("sectors", [])
|
||||
|
||||
# 代码→名称映射（优先用本地缓存，避免每次跑都调akshare）
|
||||
cache_path = DATA_DIR / "stock_name_code_cache.json"
|
||||
name_to_code = {}
|
||||
if cache_path.exists():
|
||||
name_to_code = json.loads(cache_path.read_text())
|
||||
if not name_to_code:
|
||||
try:
|
||||
import akshare as ak
|
||||
df = ak.stock_info_a_code_name()
|
||||
for _, r in df.iterrows():
|
||||
name_to_code[r["name"].strip()] = r["code"]
|
||||
cache_path.write_text(json.dumps(name_to_code, ensure_ascii=False))
|
||||
print(f" 名称代码映射: {len(name_to_code)}只已缓存", flush=True)
|
||||
except Exception as e:
|
||||
print(f" 名称代码映射加载失败: {e}", flush=True)
|
||||
|
||||
leaders = []
|
||||
seen = set()
|
||||
for s in sectors:
|
||||
chg = s.get("change", 0) or 0
|
||||
lead_name = s.get("lead_stock", "")
|
||||
if chg < SECTOR_HOT_THRESHOLD or not lead_name or lead_name in seen:
|
||||
continue
|
||||
seen.add(lead_name)
|
||||
code = name_to_code.get(lead_name, "")
|
||||
if not code:
|
||||
continue
|
||||
leaders.append({
|
||||
"code": code,
|
||||
"name": lead_name,
|
||||
"source": f"行业领涨-{s['name']}+{chg:+.1f}%",
|
||||
})
|
||||
return leaders
|
||||
except Exception as e:
|
||||
print(f" 行业领涨获取失败: {e}", flush=True)
|
||||
return []
|
||||
|
||||
|
||||
def get_scanned_codes(conn):
|
||||
"""取1小时内已扫描过的代码"""
|
||||
rows = conn.execute(
|
||||
"SELECT code FROM xiaoguo_scan_tracker WHERE datetime(last_scanned_at) > datetime('now', '-1 hour')"
|
||||
).fetchall()
|
||||
return {r[0] for r in rows}
|
||||
|
||||
|
||||
def mark_scanned(conn, code, name, found):
|
||||
conn.execute(
|
||||
"INSERT OR REPLACE INTO xiaoguo_scan_tracker (code, name, last_scanned_at, found_count) "
|
||||
"VALUES (?, ?, datetime('now','localtime'), COALESCE((SELECT found_count FROM xiaoguo_scan_tracker WHERE code=?),0)+?)",
|
||||
(code, name, code, 1 if found else 0)
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
|
||||
def search_news(code, max_results=3):
|
||||
"""akshare搜个股新闻"""
|
||||
articles = []
|
||||
if not HAS_AKSHARE:
|
||||
return articles
|
||||
try:
|
||||
clean_proxy()
|
||||
df = ak.stock_news_em(symbol=code)
|
||||
for _, r in df.head(max_results).iterrows():
|
||||
title = r.get('新闻标题', '')
|
||||
content = r.get('新闻内容', '')
|
||||
if title and len(title) > 5:
|
||||
articles.append({"title": title, "content": content})
|
||||
except:
|
||||
pass
|
||||
return articles
|
||||
|
||||
|
||||
def check_stock(code, name, articles):
|
||||
"""小果LLM判断这只股票是否有料（一次调用判断所有文章）"""
|
||||
if not articles:
|
||||
return None, None
|
||||
|
||||
lines = [f"{i+1}. {a['title']}" for i, a in enumerate(articles[:3])]
|
||||
prompt = f"""以下是最新关于{name}({code})的新闻标题。
|
||||
该股今日上了人气热榜/技术榜单。
|
||||
|
||||
新闻：
|
||||
{chr(10).join(lines)}
|
||||
|
||||
这只股上榜是否跟这些新闻有关？有关的话是利好还是利空？
|
||||
回答格式：有关（利好|利空|中性） 或 无关
|
||||
回答："""
|
||||
|
||||
payload = json.dumps({
|
||||
"model": XIAOGUO_MODEL,
|
||||
"messages": [{"role": "user", "content": prompt}],
|
||||
"temperature": 0.1, "max_tokens": 100,
|
||||
}).encode()
|
||||
|
||||
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
||||
req = urllib.request.Request(XIAOGUO_API, data=payload,
|
||||
headers={"Content-Type": "application/json"}, method="POST")
|
||||
try:
|
||||
resp = opener.open(req, timeout=30)
|
||||
reply = json.loads(resp.read())["choices"][0]["message"]["content"]
|
||||
if "有关" in reply or "利好" in reply or "利空" in reply:
|
||||
for s in ["利好", "利空", "中性"]:
|
||||
if s in reply:
|
||||
return True, s
|
||||
return True, "中性"
|
||||
except Exception as e:
|
||||
# LLM不可达 → 降级：标记为unknown，不阻塞扫描流程
|
||||
print(f" ⚠️ 小果LLM不可达({str(e)[:30]})，降级为unknown", flush=True)
|
||||
return True, "unknown"
|
||||
return None, None
|
||||
|
||||
|
||||
def main():
|
||||
start_time = time.time()
|
||||
conn = get_conn()
|
||||
|
||||
# 1. 拉板
|
||||
hot = fetch_hot_board()
|
||||
rotating, is_bullish = fetch_rotating_board()
|
||||
leaders = fetch_sector_leaders()
|
||||
elapsed = time.time() - start_time
|
||||
print(f"榜单: 东方财富{len(hot)}只, 同花顺{len(rotating)}只, 行业领涨{len(leaders)}只 ({elapsed:.0f}s)", flush=True)
|
||||
|
||||
if not hot and not rotating and not leaders:
|
||||
conn.close()
|
||||
return
|
||||
|
||||
# 加载持仓代码（用于看空榜比对）
|
||||
holdings = set()
|
||||
if not is_bullish:
|
||||
cur = conn.execute("SELECT code FROM holdings WHERE is_active=1")
|
||||
holdings = {r[0].lstrip("0") for r in cur.fetchall()}
|
||||
# 也查自选
|
||||
cur2 = conn.execute("SELECT code FROM watchlist_stocks")
|
||||
holdings.update({r[0].lstrip("0") for r in cur2.fetchall()})
|
||||
|
||||
# 2. 合并去重 + 看空榜只保留持仓股
|
||||
all_stocks = {}
|
||||
# 行业领涨优先（热门板块龙头）
|
||||
for s in (leaders if is_bullish else []) + hot + rotating:
|
||||
code = s["code"]
|
||||
code_stripped = code.lstrip("0")
|
||||
if not is_bullish:
|
||||
# 看空榜：只处理持仓/自选中的股票
|
||||
if code_stripped not in holdings:
|
||||
continue
|
||||
if code not in all_stocks:
|
||||
all_stocks[code] = {"code": code, "name": s["name"], "sources": []}
|
||||
all_stocks[code]["sources"].append(s["source"])
|
||||
|
||||
if not all_stocks:
|
||||
if is_bullish:
|
||||
print("榜单为空", flush=True)
|
||||
else:
|
||||
print(f"看空榜无持仓股命中", flush=True)
|
||||
conn.close()
|
||||
return
|
||||
|
||||
# 3. 排除已搜索过的（看空榜不排除——每次都要检查风险）
|
||||
scanned = get_scanned_codes(conn)
|
||||
if is_bullish:
|
||||
candidates = [s for code, s in all_stocks.items()
|
||||
if code not in scanned and len(code) == 6 and code.isdigit()][:MAX_STOCKS_PER_RUN]
|
||||
else:
|
||||
# 看空榜：不限数量，全检
|
||||
candidates = [s for code, s in all_stocks.items()
|
||||
if len(code) == 6 and code.isdigit()]
|
||||
|
||||
if not candidates:
|
||||
print(f"无新候选（已有 {len(scanned)} 只已扫描）", flush=True)
|
||||
conn.close()
|
||||
return
|
||||
|
||||
print(f"待扫描: {len(candidates)} 只（{'看多' if is_bullish else '看空'}榜）", flush=True)
|
||||
|
||||
# 4. 逐只处理
|
||||
found_any = False
|
||||
for stock in candidates:
|
||||
code, name = stock["code"], stock["name"]
|
||||
sources = "|".join(stock["sources"])
|
||||
|
||||
articles = search_news(code, ARTICLES_PER_STOCK) if is_bullish else []
|
||||
if not articles and is_bullish:
|
||||
mark_scanned(conn, code, name, False)
|
||||
continue
|
||||
|
||||
has_found = False
|
||||
if is_bullish:
|
||||
ok, sentiment = check_stock(code, name, articles)
|
||||
if ok:
|
||||
has_found = True
|
||||
found_any = True
|
||||
conn.execute(
|
||||
"INSERT INTO signal_news (signal_id, sector, overall_sentiment, summary, key_articles, searched_stocks, source) "
|
||||
"VALUES (NULL, ?, ?, ?, ?, ?, 'xiaoguo')",
|
||||
(f"扫描-{name}", sentiment, f"[{sources}] {articles[0]['title'][:80]}",
|
||||
json.dumps([{"title": a["title"], "sentiment": sentiment, "summary": (a.get("content") or "")[:100]} for a in articles[:3]], ensure_ascii=False),
|
||||
json.dumps([code, name], ensure_ascii=False))
|
||||
)
|
||||
print(f" ✅ {name}({code}) [{sources}] {sentiment}: {articles[0]['title'][:50]}", flush=True)
|
||||
mark_scanned(conn, code, name, has_found)
|
||||
else:
|
||||
# 看空榜：直接写入风险信号
|
||||
has_found = True
|
||||
found_any = True
|
||||
conn.execute(
|
||||
"INSERT INTO signal_news (signal_id, sector, overall_sentiment, summary, key_articles, searched_stocks, source) "
|
||||
"VALUES (NULL, ?, ?, ?, ?, ?, 'xiaoguo_risk')",
|
||||
(f"预警-{name}", "偏空", f"[{sources}] {name}登上{sources}榜，需关注持仓风险",
|
||||
json.dumps([{"title": name, "sentiment": "偏空", "summary": f"上榜{sources}"}], ensure_ascii=False),
|
||||
json.dumps([code, name], ensure_ascii=False))
|
||||
)
|
||||
print(f" ⚠️ {name}({code}) [{sources}] 持仓风险信号", flush=True)
|
||||
mark_scanned(conn, code, name, has_found)
|
||||
|
||||
total_time = time.time() - start_time
|
||||
print(f"完成: {len(candidates)}只{'看多' if is_bullish else '看空'}扫描, {'有发现' if found_any else '无发现'} ({total_time:.0f}s)", flush=True)
|
||||
conn.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,74 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""xiaoguo_sentiment_bridge.py — 小果情感分析数据 → 策略引擎桥接
|
||||
|
||||
小果情感分析 cron (3da7ad4ff3a6) 每日16:00运行，输出到指定格式。
|
||||
本脚本读取小果的输出，格式化为 strategy_lifecycle 可读取的格式，
|
||||
写入 /home/hmo/web-dashboard/data/xiaoguo_sentiment.json
|
||||
|
||||
格式：
|
||||
{
|
||||
"updated_at": "2026-06-18T16:00:00",
|
||||
"stocks": {
|
||||
"00700": {
|
||||
"name": "腾讯控股",
|
||||
"sentiment": "positive" | "negative" | "neutral",
|
||||
"confidence": 0.85,
|
||||
"keywords": ["游戏", "增长"],
|
||||
"summary": "腾讯游戏业务Q2增长超预期",
|
||||
"source": "xiaoguo"
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from datetime import datetime
|
||||
|
||||
OUTPUT_PATH = "/home/hmo/web-dashboard/data/xiaoguo_sentiment.json"
|
||||
XIAOGUO_INSIGHTS_PATH = "/home/hmo/web-dashboard/data/xiaoguo_insights.json"
|
||||
|
||||
|
||||
def load_xiaoguo_output():
|
||||
"""读取小果情感分析的最新输出"""
|
||||
try:
|
||||
if os.path.exists(XIAOGUO_INSIGHTS_PATH):
|
||||
with open(XIAOGUO_INSIGHTS_PATH) as f:
|
||||
data = json.load(f)
|
||||
return data
|
||||
except Exception:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
def main():
|
||||
data = load_xiaoguo_output()
|
||||
if data is None:
|
||||
data = {"updated_at": datetime.now().isoformat(), "stocks": {}}
|
||||
else:
|
||||
# 转换为 {code: {sentiment, confidence, keywords, summary}} 格式
|
||||
formatted = {"updated_at": datetime.now().isoformat(), "stocks": {}}
|
||||
for item in data.get("analyses", []):
|
||||
code = item.get("code", "")
|
||||
if code:
|
||||
formatted["stocks"][code] = {
|
||||
"name": item.get("name", ""),
|
||||
"sentiment": item.get("sentiment", "neutral"),
|
||||
"confidence": item.get("confidence", 0),
|
||||
"keywords": item.get("keywords", []),
|
||||
"summary": item.get("brief", ""),
|
||||
"source": "xiaoguo",
|
||||
}
|
||||
data = formatted
|
||||
|
||||
os.makedirs(os.path.dirname(OUTPUT_PATH), exist_ok=True)
|
||||
with open(OUTPUT_PATH, "w") as f:
|
||||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||||
|
||||
stock_count = len(data.get("stocks", {}))
|
||||
print(f"[xiaoguo_bridge] {stock_count} stocks synced", file=sys.stderr)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,298 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
"""xiaoguo_signal_consumer.py — 知微消费小果扫描信号
|
||||
|
||||
盘中每30分钟运行，读取 signal_news 表中未处理的 xiaoguo 信号，
|
||||
做五维快速评估后决定：加自选 / 关注 / 跳过。
|
||||
|
||||
管道位置：
|
||||
xiaoguo_scanner (每5分) → signal_news → 本脚本 → 知微分析报告
|
||||
|
||||
no_agent模式：有发现→输出，无→静默
|
||||
"""
|
||||
|
||||
import json, os, sqlite3, sys, time, urllib.request
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
# 确保 MoFin 根目录在模块搜索路径中（兼容 cron 环境）
|
||||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||||
|
||||
from mo_data import read_watchlist
|
||||
from mofin_db import write_watchlist_stock
|
||||
|
||||
BASE = Path("/home/hmo/MoFin")
|
||||
DATA = BASE / "data"
|
||||
DB_PATH = DATA / "mofin.db"
|
||||
|
||||
SIGNAL_MAX_AGE_HOURS = 4 # 只处理4小时内产生的信号
|
||||
|
||||
|
||||
def clean_proxy():
|
||||
for k in ['http_proxy','https_proxy','HTTP_PROXY','HTTPS_PROXY']:
|
||||
os.environ.pop(k, None)
|
||||
|
||||
|
||||
def fetch_quote(code):
|
||||
"""拉行情。DB 优先，腾讯 API fallback"""
|
||||
# DB 优先
|
||||
try:
|
||||
from mofin_db import get_price_from_db
|
||||
p, chg = get_price_from_db(code)
|
||||
if p:
|
||||
return {"name":"", "code":code, "price":p, "change_pct":chg or 0}
|
||||
except:
|
||||
pass
|
||||
# Fallback: 腾讯实时行情 API
|
||||
try:
|
||||
url = f"http://qt.gtimg.cn/q={code}"
|
||||
resp = urllib.request.urlopen(url, timeout=10).read().decode("gbk")
|
||||
parts = resp.split("~")
|
||||
if len(parts) > 32:
|
||||
name = parts[1]
|
||||
price = float(parts[3]) if parts[3] else None
|
||||
chg_pct = float(parts[32]) if parts[32] else 0
|
||||
if price:
|
||||
return {"name": name, "code": code, "price": price, "change_pct": chg_pct, "pe": 0, "turnover": 0}
|
||||
return {"code": code, "error": "取价失败"}
|
||||
except Exception as e:
|
||||
return {"code": code, "error": str(e)[:60]}
|
||||
|
||||
|
||||
def is_in_portfolio(conn, code):
|
||||
"""检查是否已在持仓或自选中"""
|
||||
code_stripped = code.lstrip("0")
|
||||
cur = conn.execute("SELECT COUNT(*) FROM holdings WHERE code=? AND is_active=1", (code_stripped,))
|
||||
if cur.fetchone()[0] > 0:
|
||||
return "holdings"
|
||||
cur = conn.execute("SELECT COUNT(*) FROM watchlist_stocks WHERE code=?", (code,))
|
||||
if cur.fetchone()[0] > 0:
|
||||
return "watchlist"
|
||||
# 也检查 watchlist.json
|
||||
try:
|
||||
wl = read_watchlist()
|
||||
for s in wl.get("stocks", []):
|
||||
if s.get("code") == code or s.get("code", "").lstrip("0") == code_stripped:
|
||||
return "watchlist"
|
||||
except:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
def quick_assess(quote):
|
||||
"""五维快速评估（自动版）"""
|
||||
score = 0
|
||||
reasons = []
|
||||
|
||||
# 大盘环境，从DB读（回退JSON）
|
||||
try:
|
||||
import sqlite3
|
||||
conn = sqlite3.connect(str(DB_PATH))
|
||||
row = conn.execute(
|
||||
"SELECT indices FROM macro_context_log WHERE has_valid_data=1 ORDER BY created_at DESC LIMIT 1"
|
||||
).fetchone()
|
||||
conn.close()
|
||||
if row and row[0]:
|
||||
mc = json.loads(row[0])
|
||||
else:
|
||||
raise ValueError
|
||||
sh = 0
|
||||
for k, v in mc.items():
|
||||
if "上证" in k:
|
||||
sh = v.get("change_pct", 0)
|
||||
break
|
||||
if sh > 0.5:
|
||||
score += 1
|
||||
reasons.append(f"大盘+{sh:.1f}%偏强")
|
||||
elif sh < -0.5:
|
||||
score -= 1
|
||||
reasons.append(f"大盘{sh:.1f}%偏弱")
|
||||
except Exception:
|
||||
try:
|
||||
mc = json.loads((DATA / "macro_context.json").read_text())
|
||||
sh = mc.get("shanghai", {}).get("change_pct", 0)
|
||||
if sh > 0.5:
|
||||
score += 1
|
||||
reasons.append(f"大盘+{sh:.1f}%偏强")
|
||||
elif sh < -0.5:
|
||||
score -= 1
|
||||
reasons.append(f"大盘{sh:.1f}%偏弱")
|
||||
except:
|
||||
pass
|
||||
|
||||
# 技术面：涨跌幅
|
||||
chg = quote.get("change_pct", 0)
|
||||
if chg > 3:
|
||||
score += 1
|
||||
reasons.append(f"涨幅+{chg:.1f}%偏强")
|
||||
elif chg < -3:
|
||||
score -= 1
|
||||
reasons.append(f"跌幅{chg:.1f}%偏弱")
|
||||
else:
|
||||
score += 0.5
|
||||
reasons.append(f"走势平稳{chg:+.1f}%")
|
||||
|
||||
# 基本面：PE
|
||||
pe = quote.get("pe", 0)
|
||||
if 5 < pe < 40:
|
||||
score += 1
|
||||
reasons.append(f"PE={pe:.0f}合理")
|
||||
elif pe <= 0:
|
||||
score -= 0.5
|
||||
reasons.append("PE为负")
|
||||
elif pe > 100:
|
||||
score -= 0.5
|
||||
reasons.append(f"PE={pe:.0f}偏高")
|
||||
|
||||
# 量能
|
||||
turn = quote.get("turnover", 0)
|
||||
if turn > 5:
|
||||
score += 0.5
|
||||
reasons.append(f"换手{turn:.1f}%活跃")
|
||||
elif turn < 0.5:
|
||||
score -= 0.3
|
||||
reasons.append(f"换手{turn:.1f}%偏低")
|
||||
|
||||
return score, reasons
|
||||
|
||||
|
||||
def evaluate_and_act(signal, quote):
|
||||
"""评估信号并决定操作"""
|
||||
status_in = is_in_portfolio(get_conn(), signal.get("code", ""))
|
||||
if status_in:
|
||||
return f"已在{status_in}中，跳过", None
|
||||
|
||||
score, reasons = quick_assess(quote)
|
||||
|
||||
if score >= 1.5:
|
||||
action = "watchlist"
|
||||
summary = f"加自选: {' | '.join(reasons)}"
|
||||
elif score >= 0:
|
||||
action = "monitor"
|
||||
summary = f"关注: {' | '.join(reasons)}"
|
||||
else:
|
||||
action = "skip"
|
||||
summary = f"跳过(评分{score:.1f}): {' | '.join(reasons)}"
|
||||
|
||||
return summary, action
|
||||
|
||||
|
||||
def get_conn():
|
||||
import sqlite3
|
||||
conn = sqlite3.connect(str(DB_PATH))
|
||||
conn.row_factory = sqlite3.Row
|
||||
return conn
|
||||
|
||||
|
||||
def mark_processed(conn, signal_id):
|
||||
conn.execute("UPDATE signal_news SET processed=1 WHERE id=?", (signal_id,))
|
||||
conn.commit()
|
||||
|
||||
|
||||
def main():
|
||||
clean_proxy()
|
||||
start = time.time()
|
||||
today = datetime.now().strftime("%Y-%m-%d")
|
||||
|
||||
conn = get_conn()
|
||||
|
||||
# 读未处理 xiaoguo 信号（SIGNAL_MAX_AGE_HOURS 以内）
|
||||
rows = conn.execute(
|
||||
"SELECT id, sector, overall_sentiment, summary, key_articles, searched_stocks, source "
|
||||
"FROM signal_news "
|
||||
"WHERE source LIKE 'xiaoguo%' AND (processed=0 OR processed IS NULL) "
|
||||
f"AND created_at > datetime('now', '-{SIGNAL_MAX_AGE_HOURS} hours') "
|
||||
"ORDER BY created_at DESC LIMIT 20"
|
||||
).fetchall()
|
||||
|
||||
if not rows:
|
||||
# 标记过期信号为已处理（超出时效边界）
|
||||
old = conn.execute(f"SELECT COUNT(*) FROM signal_news WHERE source LIKE 'xiaoguo%' AND (processed=0 OR processed IS NULL) AND created_at <= datetime('now', '-{SIGNAL_MAX_AGE_HOURS} hours')").fetchone()[0]
|
||||
if old:
|
||||
conn.execute(f"UPDATE signal_news SET processed=1 WHERE source LIKE 'xiaoguo%' AND (processed=0 OR processed IS NULL) AND created_at <= datetime('now', '-{SIGNAL_MAX_AGE_HOURS} hours')")
|
||||
conn.commit()
|
||||
print(f"[SILENT] 清理 {old} 条过期信号（>{SIGNAL_MAX_AGE_HOURS}h）")
|
||||
else:
|
||||
print("[SILENT] 今日无未处理小果信号")
|
||||
conn.close()
|
||||
return
|
||||
|
||||
# 尝试从 searched_stocks 提取股票代码
|
||||
results = []
|
||||
for r in rows:
|
||||
try:
|
||||
searched = json.loads(r["searched_stocks"]) if r["searched_stocks"] else []
|
||||
except:
|
||||
searched = []
|
||||
|
||||
# 从 sector 字段取股票名
|
||||
sector_name = r["sector"] or ""
|
||||
|
||||
# 尝试提取代码
|
||||
codes_found = []
|
||||
for s in searched:
|
||||
# searched_stocks 存的是股票名称列表
|
||||
# 尝试从 summary 里找代码
|
||||
import re
|
||||
codes = re.findall(r'\d{6}', r["summary"] or "")
|
||||
codes_found.extend(codes)
|
||||
|
||||
if not codes_found:
|
||||
# 没有直接代码，用名称去查
|
||||
mark_processed(conn, r["id"])
|
||||
continue
|
||||
|
||||
code = codes_found[0]
|
||||
quote = fetch_quote(code)
|
||||
|
||||
summary, action = evaluate_and_act(dict(r), quote)
|
||||
|
||||
if action == "watchlist":
|
||||
# 加自选
|
||||
results.append(f"✅ {sector_name}({code}): {summary}")
|
||||
# 写入 watchlist_stocks 表(DB)
|
||||
try:
|
||||
wl = read_watchlist()
|
||||
wl.setdefault("stocks", [])
|
||||
# 检查是否已在
|
||||
existing = [s for s in wl["stocks"] if s.get("code") == code]
|
||||
if not existing:
|
||||
new_stock = {
|
||||
"code": code,
|
||||
"name": quote.get("name", sector_name),
|
||||
"price": quote.get("price", 0),
|
||||
"status": "watching",
|
||||
"source": "xiaoguo_scanner",
|
||||
"added_at": today,
|
||||
}
|
||||
wl["stocks"].append(new_stock)
|
||||
# DB 写入（watchlist_stocks）
|
||||
try:
|
||||
conn2 = get_conn()
|
||||
new_stock["currency"] = "CNY"
|
||||
write_watchlist_stock(conn2, new_stock)
|
||||
conn2.close()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
except:
|
||||
pass
|
||||
elif action == "monitor":
|
||||
results.append(f"🔄 {sector_name}({code}): {summary}")
|
||||
else:
|
||||
results.append(f"⏭️ {sector_name}({code}): {summary}")
|
||||
|
||||
mark_processed(conn, r["id"])
|
||||
|
||||
conn.close()
|
||||
|
||||
elapsed = time.time() - start
|
||||
if results:
|
||||
print(f"小果信号消费 | {today} | {len(results)}条处理 ({elapsed:.0f}s)")
|
||||
for r in results:
|
||||
print(f" {r}")
|
||||
else:
|
||||
print("[SILENT] 小果信号消费结束")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user