104 lines
3.8 KiB
Python
104 lines
3.8 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""leader_scanner.py — MoFin 龙头识别策略（bull_trend 适用，2026-08-13 落地）
|
||
|
||
核心逻辑（12维方法论 + 龙头识别低波动优化版理念）：
|
||
- 适用状态：bull_trend（above_ma20 + rsi>55 + adx>20）
|
||
- 入场：龙头股回调到 MA20 附近（趋势中的健康回调，不追高）
|
||
- 12维条件（复用 stock_indicators 已有字段，无拍脑袋）：
|
||
1. 个股 close > MA20（趋势向上）
|
||
2. dist_ma20 ∈ [-3%, +2%]（回调到 MA20 附近，不追高）
|
||
3. 行业强势（sector_above_ma20=1 或 sector_ret20>0，需 sector 数据）
|
||
4. 市值/流动性过滤（mcap_q > 0.3，龙头非小盘）
|
||
5. RSI ∈ [45, 65]（强势但未超买）
|
||
6. 量能配合（vol_ratio > 0.8，非极度缩量）
|
||
- 出场：跌破 MA20 或达到上方压力位
|
||
- RR：压力位（前高/筹码阻力）/ 支撑位（MA20）计算
|
||
|
||
注意：这是初版框架，需 MoFin 引擎验证达标（年化≥13.2%）才正式上线
|
||
"""
|
||
import sqlite3
|
||
import json
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
DB = "/home/hmo/MoFin/data/mofin.db"
|
||
OUT = Path("/home/hmo/MoFin/data/leader_signals.json")
|
||
|
||
def load_market_state():
|
||
p = Path("/home/hmo/MoFin/data/market_state.json")
|
||
if p.exists():
|
||
return json.loads(p.read_text(encoding="utf-8"))
|
||
return {"state": "neutral"}
|
||
|
||
def scan_leaders():
|
||
"""扫描龙头股回调买点（bull_trend 状态）"""
|
||
ms = load_market_state()
|
||
state = ms.get("state", "neutral")
|
||
if state != "bull_trend":
|
||
print(f"当前状态 {state}，非 bull_trend，不扫描龙头（避免追高）")
|
||
return []
|
||
|
||
c = sqlite3.connect(DB)
|
||
# 最新交易日
|
||
row = c.execute("SELECT MAX(date) FROM stock_indicators").fetchone()
|
||
if not row or not row[0]:
|
||
c.close()
|
||
return []
|
||
latest = row[0]
|
||
print(f"扫描日期: {latest}")
|
||
|
||
# 龙头条件（12维）
|
||
rows = c.execute(
|
||
"""SELECT code, ma20, rsi, dist_ma20, mcap_q, pe_q, vol_ratio, trend_aligned
|
||
FROM stock_indicators
|
||
WHERE date=? AND ma20 IS NOT NULL AND rsi IS NOT NULL""",
|
||
(latest,)
|
||
).fetchall()
|
||
c.close()
|
||
|
||
signals = []
|
||
for r in rows:
|
||
code, ma20, rsi, dist_ma20, mcap_q, pe_q, vol_ratio, trend_aligned = r
|
||
# 条件1: 趋势向上（close > MA20 → dist_ma20 > 0，或接近）
|
||
if dist_ma20 is None or dist_ma20 < -3 or dist_ma20 > 2:
|
||
continue
|
||
# 条件2: RSI 强势未超买
|
||
if rsi < 45 or rsi > 65:
|
||
continue
|
||
# 条件3: 市值/流动性（龙头非小盘，mcap_q > 0.3）
|
||
if mcap_q is not None and mcap_q < 0.3:
|
||
continue
|
||
# 条件4: 量能配合
|
||
if vol_ratio is not None and vol_ratio < 0.8:
|
||
continue
|
||
# 条件5: 趋势共振（trend_aligned=1）
|
||
if trend_aligned != 1:
|
||
continue
|
||
signals.append({
|
||
"code": code, "ma20": ma20, "rsi": rsi,
|
||
"dist_ma20": dist_ma20, "mcap_q": mcap_q, "pe_q": pe_q,
|
||
"vol_ratio": vol_ratio, "entry_reason": "龙头回调MA20",
|
||
})
|
||
|
||
# 按 dist_ma20 排序（最接近 MA20 的优先）
|
||
signals.sort(key=lambda x: abs(x["dist_ma20"]))
|
||
print(f"龙头信号: {len(signals)} 只")
|
||
for s in signals[:5]:
|
||
print(f" {s['code']}: dist_ma20={s['dist_ma20']:.1f}% rsi={s['rsi']:.1f} mcap_q={s['mcap_q']}")
|
||
return signals
|
||
|
||
def main():
|
||
signals = scan_leaders()
|
||
out = {
|
||
"state": load_market_state().get("state", "neutral"),
|
||
"signals": signals,
|
||
"count": len(signals),
|
||
"updated_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
}
|
||
OUT.write_text(json.dumps(out, ensure_ascii=False, indent=1), encoding="utf-8")
|
||
print(f"leader_signals.json 写入")
|
||
|
||
if __name__ == "__main__":
|
||
main()
|