feat: 宏观指数采集独立(macro_index_collector) + 莫荷宏观预警系统入库
- 重建 macro_index_collector.py: 原macro_context_collector(index采集)独立 避免与莫荷的 macro_context_collector(宏观新闻采集)冲突 - 纳入莫荷的宏观预警系统: divergence_detector.py — 跨市场背离监测 macro_context_collector.py — 宏观新闻采集+红绿灯(莫荷版) macro_signal_consumer.py — 宏观风险信号消费 - cron修正: 宏观采集-早/午间 → macro_index_collector.py
This commit is contained in:
知微
@@ -0,0 +1,333 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
divergence_detector.py — 跨市场背离监测器（no_agent）
|
||||
|
||||
每30分钟检测：
|
||||
1. 科创50 vs 恒指 → 科技股超买/超卖信号
|
||||
2. A/H 隐含溢价 → 内外资分歧度
|
||||
3. 上证50 vs 创业板 → 风格轮动信号
|
||||
4. 恒指 vs 国企指数 → 离岸市场情绪
|
||||
5. 指数连涨/连跌天数 → 趋势延续/衰竭
|
||||
|
||||
输出：
|
||||
- HIGH/medium divergence → 写入 signal_news (source=divergence_watch)
|
||||
- 状态文件 macro_divergence_state.json
|
||||
- no_agent: 有信号才出声
|
||||
"""
|
||||
import sys, json, re, datetime, os, urllib.request
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
BASE = Path("/home/hmo/MoFin")
|
||||
DATA = BASE / "data"
|
||||
DB_PATH = DATA / "mofin.db"
|
||||
STATE_PATH = DATA / "macro_divergence_state.json"
|
||||
|
||||
# ── 监测的指数 ──
|
||||
INDEX_CODES = {
|
||||
"上证指数": "sh000001",
|
||||
"深证成指": "sz399001",
|
||||
"创业板指": "sz399006",
|
||||
"科创50": "sh000688",
|
||||
"上证50": "sh000016",
|
||||
"沪深300": "sh000300",
|
||||
"恒生指数": "hkHSI",
|
||||
"国企指数": "hkHSCEI",
|
||||
}
|
||||
|
||||
# ── 背离阈值 ──
|
||||
DIVERGENCE_STRONG = 5.0 # >5% → strong信号
|
||||
DIVERGENCE_MODERATE = 3.0 # >3% → moderate信号
|
||||
STREAK_DAYS = 3 # 连涨/连跌3天 → 信号
|
||||
|
||||
def fetch_indices():
|
||||
"""获取所有指数实时数据"""
|
||||
symbols = list(INDEX_CODES.values())
|
||||
url = f"http://qt.gtimg.cn/q={','.join(symbols)}"
|
||||
try:
|
||||
r = urllib.request.urlopen(url, timeout=10)
|
||||
text = r.read().decode("gbk")
|
||||
except Exception as e:
|
||||
print(f"[DIVERGE] 采集失败: {e}", file=sys.stderr)
|
||||
return {}
|
||||
|
||||
indices = {}
|
||||
for line in text.strip().split("\n"):
|
||||
line = line.strip()
|
||||
if not line or "=" not in line:
|
||||
continue
|
||||
try:
|
||||
sym = line.split("=", 1)[0].strip().lstrip("v_")
|
||||
raw = line.split("=", 1)[1].strip().strip('"').strip(";")
|
||||
fields = raw.split("~")
|
||||
if len(fields) < 35:
|
||||
continue
|
||||
name = fields[1]
|
||||
price = float(fields[3]) if fields[3].strip() else 0
|
||||
close = float(fields[4]) if fields[4].strip() else 0
|
||||
change_pct = ((price - close) / close * 100) if close else 0
|
||||
high = float(fields[33]) if fields[33].strip() else 0
|
||||
low = float(fields[34]) if fields[34].strip() else 0
|
||||
timestamp = fields[30] if len(fields) > 30 else ""
|
||||
indices[sym] = {
|
||||
"name": name,
|
||||
"price": price,
|
||||
"close": close,
|
||||
"change_pct": round(change_pct, 2),
|
||||
"high": high,
|
||||
"low": low,
|
||||
"timestamp": timestamp,
|
||||
}
|
||||
except Exception:
|
||||
continue
|
||||
return indices
|
||||
|
||||
def load_history():
|
||||
"""从MACRO_CONTEXT_LOG加载前几天的指数数据"""
|
||||
try:
|
||||
import sqlite3
|
||||
conn = sqlite3.connect(str(DB_PATH))
|
||||
rows = conn.execute(
|
||||
"SELECT indices, created_at FROM macro_context_log "
|
||||
"WHERE has_valid_data=1 ORDER BY created_at DESC LIMIT 5"
|
||||
).fetchall()
|
||||
conn.close()
|
||||
history = []
|
||||
for row in rows:
|
||||
idx_data = json.loads(row[0]) if row[0] else {}
|
||||
history.append({
|
||||
"indices": idx_data,
|
||||
"timestamp": row[1],
|
||||
})
|
||||
return history
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
def detect_divergences(indices, history):
|
||||
"""
|
||||
检测跨市场背离信号
|
||||
返回: list of signal dicts {type, level, desc, pairs}
|
||||
"""
|
||||
signals = []
|
||||
|
||||
def get(name):
|
||||
"""按中文名找指数"""
|
||||
sym = INDEX_CODES.get(name)
|
||||
if sym and sym in indices:
|
||||
return indices[sym]
|
||||
# 模糊匹配
|
||||
for s, idx in indices.items():
|
||||
if name in idx.get("name", ""):
|
||||
return idx
|
||||
return None
|
||||
|
||||
sh_comp = get("上证指数")
|
||||
sz_comp = get("深证成指")
|
||||
cyb = get("创业板指")
|
||||
kc = get("科创50")
|
||||
sz50 = get("上证50")
|
||||
hs300 = get("沪深300")
|
||||
hsi = get("恒生指数")
|
||||
hscei = get("国企指数")
|
||||
|
||||
# ── 信号1: 科创50 vs 恒指（科技股vs国际资本）──
|
||||
if kc and hsi:
|
||||
kc_chg = kc["change_pct"]
|
||||
hsi_chg = hsi["change_pct"]
|
||||
divergence = abs(kc_chg - hsi_chg)
|
||||
if divergence > DIVERGENCE_STRONG:
|
||||
signals.append({
|
||||
"type": "a_h_tech_divergence",
|
||||
"level": "high",
|
||||
"desc": f"科创50({kc_chg:+.1f}%) vs 恒指({hsi_chg:+.1f}%) 背离{divergence:.1f}个百分点",
|
||||
"pairs": [kc, hsi],
|
||||
"direction": "risk" if kc_chg > hsi_chg else "opportunity",
|
||||
})
|
||||
elif divergence > DIVERGENCE_MODERATE:
|
||||
signals.append({
|
||||
"type": "a_h_tech_divergence",
|
||||
"level": "medium",
|
||||
"desc": f"科创50({kc_chg:+.1f}%) vs 恒指({hsi_chg:+.1f}%) 背离{divergence:.1f}个百分点",
|
||||
"pairs": [kc, hsi],
|
||||
"direction": "risk" if kc_chg > hsi_chg else "opportunity",
|
||||
})
|
||||
|
||||
# ── 信号2: 上证50 vs 创业板（价值vs成长）──
|
||||
if sz50 and cyb:
|
||||
sz50_chg = sz50["change_pct"]
|
||||
cyb_chg = cyb["change_pct"]
|
||||
divergence = abs(sz50_chg - cyb_chg)
|
||||
if divergence > DIVERGENCE_STRONG:
|
||||
direction = "opportunity" if sz50_chg > cyb_chg else "risk"
|
||||
signals.append({
|
||||
"type": "value_growth_divergence",
|
||||
"level": "high",
|
||||
"desc": f"上证50({sz50_chg:+.1f}%) vs 创业板({cyb_chg:+.1f}%) 背离{divergence:.1f}个百分点",
|
||||
"pairs": [sz50, cyb],
|
||||
"direction": direction,
|
||||
})
|
||||
elif divergence > DIVERGENCE_MODERATE:
|
||||
direction = "opportunity" if sz50_chg > cyb_chg else "risk"
|
||||
signals.append({
|
||||
"type": "value_growth_divergence",
|
||||
"level": "medium",
|
||||
"desc": f"上证50({sz50_chg:+.1f}%) vs 创业板({cyb_chg:+.1f}%) 背离{divergence:.1f}个百分点",
|
||||
"pairs": [sz50, cyb],
|
||||
"direction": direction,
|
||||
})
|
||||
|
||||
# ── 信号3: 恒指 vs 国企指数（国际资本流向）──
|
||||
if hsi and hscei:
|
||||
hsi_chg = hsi["change_pct"]
|
||||
hscei_chg = hscei["change_pct"]
|
||||
if hsi_chg < 0 and hscei_chg < hsi_chg:
|
||||
# 国企跌得比恒指多 → 外资恐慌性卖出H股
|
||||
signals.append({
|
||||
"type": "hk_panic_selling",
|
||||
"level": "high",
|
||||
"desc": f"国企指数({hscei_chg:+.1f}%)跌幅大于恒指({hsi_chg:+.1f}%)→外资恐慌抛售H股",
|
||||
"pairs": [hscei, hsi],
|
||||
"direction": "risk",
|
||||
})
|
||||
|
||||
# ── 信号4: A/H 价格背离（用历史数据检测趋势延续）──
|
||||
if history and len(history) >= 2:
|
||||
latest = history[0]["indices"]
|
||||
prev = history[1]["indices"]
|
||||
|
||||
# 科创50连涨检测
|
||||
if kc:
|
||||
kc_now = kc["change_pct"]
|
||||
kc_prev = prev.get("科创50", {}).get("change_pct", 0) if "科创50" in prev else 0
|
||||
kc_yest = latest.get("科创50", {}).get("change_pct", 0)
|
||||
|
||||
# 检测连涨
|
||||
if isinstance(kc_yest, (int, float)) and isinstance(kc_prev, (int, float)):
|
||||
streak = 0
|
||||
if kc_yest > 0: streak += 1
|
||||
if kc_prev > 0: streak += 1
|
||||
if kc_now > 0: streak += 1
|
||||
if streak >= STREAK_DAYS and kc_now > 0:
|
||||
signals.append({
|
||||
"type": "tech_streak",
|
||||
"level": "medium",
|
||||
"desc": f"科创50连涨{streak}日({kc_prev:+.1f}%→{kc_yest:+.1f}%→{kc_now:+.1f}%)→超买风险",
|
||||
"pairs": [kc],
|
||||
"direction": "risk",
|
||||
})
|
||||
|
||||
# ── 信号5: 大盘宽度 + 季节效应 ──
|
||||
now = datetime.datetime.now()
|
||||
is_month_end = now.day >= 25 # 月末最后一周
|
||||
is_friday = now.weekday() == 4
|
||||
|
||||
# 总体判断
|
||||
risk_count = sum(1 for s in signals if s["direction"] == "risk")
|
||||
opp_count = sum(1 for s in signals if s["direction"] == "opportunity")
|
||||
|
||||
# 月末+周五叠加 → 脆弱性增强
|
||||
if is_month_end and is_friday and risk_count >= 2:
|
||||
signals.append({
|
||||
"type": "time_window_risk",
|
||||
"level": "high",
|
||||
"desc": f"月末({now.day}日)+周五效应+{risk_count}个风险信号叠加→市场脆弱性高",
|
||||
"pairs": [],
|
||||
"direction": "risk",
|
||||
})
|
||||
elif is_month_end and risk_count >= 1:
|
||||
signals.append({
|
||||
"type": "time_window_risk",
|
||||
"level": "medium",
|
||||
"desc": f"月末窗口({now.day}日)+{risk_count}个风险信号→注意控制仓位",
|
||||
"pairs": [],
|
||||
"direction": "risk",
|
||||
})
|
||||
|
||||
return signals
|
||||
|
||||
def write_state(signals, indices):
|
||||
"""写入状态文件，供监控 cron 消费"""
|
||||
levels = [s["level"] for s in signals]
|
||||
highest = "high" if "high" in levels else ("medium" if "medium" in levels else "none")
|
||||
directions = [s["direction"] for s in signals]
|
||||
bias = "risk" if directions.count("risk") > directions.count("opportunity") else "opportunity"
|
||||
|
||||
state = {
|
||||
"level": highest,
|
||||
"bias": bias,
|
||||
"signal_count": len(signals),
|
||||
"signals": signals,
|
||||
"indices": {k: v["change_pct"] for k, v in indices.items()},
|
||||
"created_at": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||||
}
|
||||
STATE_PATH.write_text(json.dumps(state, ensure_ascii=False, indent=2))
|
||||
|
||||
def write_to_signal_news(signals):
|
||||
"""HIGH signal → signal_news"""
|
||||
if not signals:
|
||||
return
|
||||
import sqlite3
|
||||
conn = sqlite3.connect(str(DB_PATH))
|
||||
|
||||
high_signals = [s for s in signals if s["level"] == "high"]
|
||||
if high_signals:
|
||||
for s in high_signals:
|
||||
conn.execute(
|
||||
"INSERT INTO signal_news (signal_id, sector, overall_sentiment, summary, key_articles, searched_stocks, source) VALUES (?, ?, ?, ?, ?, ?, ?)",
|
||||
(0, "跨市场", f"背离-{s['direction'].upper()}", s["desc"], json.dumps(s, ensure_ascii=False), "", "divergence_watch")
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
med_signals = [s for s in signals if s["level"] == "medium"]
|
||||
if med_signals:
|
||||
summary = "\n".join([s["desc"] for s in med_signals])
|
||||
conn.execute(
|
||||
"INSERT INTO signal_news (signal_id, sector, overall_sentiment, summary, key_articles, searched_stocks, source) VALUES (?, ?, ?, ?, ?, ?, ?)",
|
||||
(0, "跨市场", "背离-MEDIUM", summary, json.dumps(med_signals, ensure_ascii=False), "", "divergence_watch")
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
conn.close()
|
||||
|
||||
def print_report(signals, indices):
|
||||
"""no_agent 输出"""
|
||||
if not signals:
|
||||
return # SILENT
|
||||
|
||||
# 有信号就输出
|
||||
high = [s for s in signals if s["level"] == "high"]
|
||||
med = [s for s in signals if s["level"] == "medium"]
|
||||
|
||||
lines = []
|
||||
if high:
|
||||
for s in high:
|
||||
icon = "\u26a0\ufe0f" if s["direction"] == "risk" else "\u2b06\ufe0f"
|
||||
lines.append(f"[DIVERGE] {icon} {s['level'].upper()} {s['type']}: {s['desc']}")
|
||||
if med:
|
||||
for s in med[:3]: # 最多3条
|
||||
icon = "\u26a0\ufe0f" if s["direction"] == "risk" else "\u2b06\ufe0f"
|
||||
lines.append(f"[DIVERGE] {icon} {s['level'].upper()} {s['type']}: {s['desc']}")
|
||||
|
||||
# 输出指数全景
|
||||
idx_line = " | ".join([f"{n}: {indices.get(s, {}).get('change_pct', 0):+.1f}%" for n, s in INDEX_CODES.items() if s in indices])
|
||||
lines.append(f"[DIVERGE] 指数全景: {idx_line}")
|
||||
|
||||
print("\n".join(lines))
|
||||
|
||||
def main():
|
||||
indices = fetch_indices()
|
||||
if not indices:
|
||||
return
|
||||
|
||||
history = load_history()
|
||||
signals = detect_divergences(indices, history)
|
||||
|
||||
# 写入 state + signal_news
|
||||
write_state(signals, indices)
|
||||
write_to_signal_news(signals)
|
||||
|
||||
# no_agent 输出
|
||||
print_report(signals, indices)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user