feat: 全市场异动扫描(替代小果)+自动提拔入自选
This commit is contained in:
知微
@@ -0,0 +1,194 @@
|
||||
#!/usr/bin/env python3
|
||||
"""market_scanner.py — 全市场异动扫描（替代小果扫描线）
|
||||
|
||||
每15分钟扫描：
|
||||
1. 板块轮动检测（从已采集的 sector_snapshots 读）
|
||||
2. 热门板块领涨股扫描（从腾讯API批量拉）
|
||||
3. 资金流向异常检测（从已有 capital_flow_cache 读）
|
||||
4. 输出候选股到 candidates 表
|
||||
|
||||
不依赖小果LLM，纯数据驱动。
|
||||
"""
|
||||
import sys, json, sqlite3, urllib.request, re, time
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
DB_PATH = Path("/home/hmo/MoFin/data/mofin.db")
|
||||
CANDIDATES_FILE = Path("/home/hmo/web-dashboard/data/candidate_pool.json")
|
||||
|
||||
UA = "Mozilla/5.0"
|
||||
|
||||
def get_conn():
|
||||
return sqlite3.connect(str(DB_PATH))
|
||||
|
||||
def fetch_qq_batch(symbols):
|
||||
"""腾讯批量行情"""
|
||||
if not symbols:
|
||||
return {}
|
||||
url = f"http://qt.gtimg.cn/q={','.join(symbols)}"
|
||||
try:
|
||||
req = urllib.request.Request(url, headers={"User-Agent": UA})
|
||||
proxy = urllib.request.ProxyHandler({})
|
||||
opener = urllib.request.build_opener(proxy)
|
||||
with opener.open(req, timeout=15) as r:
|
||||
text = r.read().decode("gbk")
|
||||
results = {}
|
||||
for line in text.strip().split("\n"):
|
||||
if "~" not in line:
|
||||
continue
|
||||
parts = line.split("~")
|
||||
if len(parts) < 40:
|
||||
continue
|
||||
m = re.search(r'_(\w+)=', parts[0])
|
||||
market = m.group(1) if m else ""
|
||||
code = parts[2]
|
||||
name = parts[1]
|
||||
price = float(parts[3]) if parts[3] else 0
|
||||
chg_pct = float(parts[32]) if parts[32] else 0
|
||||
high = float(parts[33]) if parts[33] else 0
|
||||
low = float(parts[34]) if parts[34] else 0
|
||||
volume = int(parts[6]) if parts[6] else 0
|
||||
amount = float(parts[37]) if len(parts) > 37 and parts[37] else 0
|
||||
if price > 0:
|
||||
results[code] = {"code": code, "name": name, "price": price,
|
||||
"change_pct": chg_pct, "high": high, "low": low,
|
||||
"volume": volume, "amount": amount,
|
||||
"market": "SH" if market == "sh" else "SZ" if market == "sz" else "HK"}
|
||||
return results
|
||||
except Exception as e:
|
||||
print(f"[SCANNER] 腾讯API错误: {e}", file=sys.stderr)
|
||||
return {}
|
||||
|
||||
def scan_hot_sectors():
|
||||
"""从DB读热门板块，返回板块名+领涨股"""
|
||||
conn = get_conn()
|
||||
latest = conn.execute("SELECT MAX(id) FROM market_snapshots").fetchone()[0]
|
||||
if not latest:
|
||||
conn.close()
|
||||
return []
|
||||
sectors = conn.execute("""
|
||||
SELECT name, change_pct, lead_stock, lead_stock_code, up_count, down_count
|
||||
FROM sector_snapshots WHERE snapshot_id=?
|
||||
ORDER BY change_pct DESC LIMIT 15
|
||||
""", (latest,)).fetchall()
|
||||
conn.close()
|
||||
return [{"name": s[0], "change": s[1], "lead_stock": s[2],
|
||||
"lead_code": s[3], "up": s[4], "down": s[5]} for s in sectors if s[1] > 2.0]
|
||||
|
||||
def scan_candidates():
|
||||
"""主扫描流程"""
|
||||
print(f"[SCANNER] {datetime.now().strftime('%H:%M')} 开始扫描", flush=True)
|
||||
|
||||
# 1. 热门板块领涨股
|
||||
hot = scan_hot_sectors()
|
||||
print(f" 热门板块(涨幅>2%): {len(hot)}个", flush=True)
|
||||
|
||||
candidates = {}
|
||||
|
||||
# 从热门板块拉领涨股
|
||||
lead_codes = []
|
||||
for s in hot:
|
||||
if s.get("lead_code"):
|
||||
lc = str(s["lead_code"]).strip()
|
||||
if lc and lc not in candidates:
|
||||
lead_codes.append(lc)
|
||||
candidates[lc] = {"source": f"板块:{s['name']}(+{s['change']:.1f}%)", "sector": s["name"]}
|
||||
|
||||
print(f" 领涨股待查: {len(lead_codes)}只", flush=True)
|
||||
|
||||
# 2. 腾讯API批量查行情
|
||||
symbols = []
|
||||
for c in lead_codes:
|
||||
if len(c) == 6:
|
||||
if c.startswith(("5", "6", "9")):
|
||||
symbols.append(f"sh{c}")
|
||||
else:
|
||||
symbols.append(f"sz{c}")
|
||||
else:
|
||||
symbols.append(f"hk{c}")
|
||||
|
||||
prices = fetch_qq_batch(symbols)
|
||||
print(f" 行情返回: {len(prices)}只", flush=True)
|
||||
|
||||
# 3. 评估候选
|
||||
new_candidates = []
|
||||
for code, info in prices.items():
|
||||
if code not in candidates:
|
||||
continue
|
||||
src = candidates[code]
|
||||
price = info["price"]
|
||||
chg = info["change_pct"]
|
||||
name = info["name"]
|
||||
vol = info["volume"]
|
||||
amt = info["amount"]
|
||||
|
||||
# 条件：涨幅>3%，有量
|
||||
if chg < 3.0:
|
||||
continue
|
||||
if vol <= 0:
|
||||
continue
|
||||
|
||||
score = min(10, round(3 + chg * 0.5 + (amt / 1e8 if amt > 0 else 0) * 0.1, 1))
|
||||
|
||||
entry_low = round(price * 0.95, 2)
|
||||
entry_high = round(price, 2)
|
||||
stop_loss = round(price * 0.92, 2)
|
||||
take_profit = round(price * 1.15, 2)
|
||||
|
||||
candidate = {
|
||||
"code": code,
|
||||
"name": name,
|
||||
"price": price,
|
||||
"change_pct": chg,
|
||||
"score": score,
|
||||
"entry_low": entry_low,
|
||||
"entry_high": entry_high,
|
||||
"stop_loss": stop_loss,
|
||||
"take_profit": take_profit,
|
||||
"source": src["source"],
|
||||
"sector": src.get("sector", ""),
|
||||
"reason": f"热门板块{src['sector']}领涨+{chg:.1f}%"
|
||||
}
|
||||
new_candidates.append(candidate)
|
||||
print(f" ✅ {code} {name} 价{price} (+{chg:.1f}%) 评分{score}", flush=True)
|
||||
|
||||
# 4. 写入candidates表
|
||||
if new_candidates:
|
||||
conn = get_conn()
|
||||
for c in new_candidates:
|
||||
conn.execute(
|
||||
"INSERT OR REPLACE INTO candidates (code, name, price, change_pct, score, "
|
||||
"entry_low, entry_high, stop_loss, take_profit, source, sector, reason, created_at) "
|
||||
"VALUES (?,?,?,?,?,?,?,?,?,?,?,?,datetime('now','localtime'))",
|
||||
(c["code"], c["name"], c["price"], c["change_pct"], c["score"],
|
||||
c["entry_low"], c["entry_high"], c["stop_loss"], c["take_profit"],
|
||||
c["source"], c["sector"], c["reason"])
|
||||
)
|
||||
conn.commit()
|
||||
conn.close()
|
||||
print(f" ✅ 写入{len(new_candidates)}只候选", flush=True)
|
||||
else:
|
||||
print(f" ⚪ 无新候选", flush=True)
|
||||
|
||||
# 5. 推送到Dad（只推高分）
|
||||
high_score = [c for c in new_candidates if c["score"] >= 6]
|
||||
if high_score:
|
||||
lines = ["🔍 市场扫描发现潜在机会："]
|
||||
for c in high_score[:3]:
|
||||
lines.append(
|
||||
f" {c['name']}({c['code']}) 价{c['price']:.2f}(+{c['change_pct']:.1f}%) "
|
||||
f"评分{c['score']}/10 | {c['reason']}"
|
||||
)
|
||||
msg = "\n".join(lines)
|
||||
# 推XMPP
|
||||
try:
|
||||
payload = json.dumps({"to": "hmo@yoin.fun", "body": msg, "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(f" 📨 已推送{len(high_score)}只高评分候选", flush=True)
|
||||
except Exception as e:
|
||||
print(f" ⚠️ 推送失败: {e}", file=sys.stderr)
|
||||
|
||||
if __name__ == "__main__":
|
||||
scan_candidates()
|
||||
+79
-195
@@ -1,207 +1,91 @@
|
||||
#!/usr/bin/env python3
|
||||
"""promote_candidates.py — 将候选股自动提拔入自选
|
||||
"""promote_candidates.py — 自动提拔候选股入自选
|
||||
|
||||
管道位置：
|
||||
candidates 表（候选池）→ 本脚本（价格验证+写入策略）→ holding_strategies（自选股）
|
||||
|
||||
运行时机：
|
||||
每30分钟，交易日 9:30~15:00
|
||||
|
||||
no_agent模式：有提拔→输出，无→静默
|
||||
|
||||
流程：
|
||||
1. 从 candidates 读 promoted=0 AND dropped=0
|
||||
2. 用 stock_quote.py 验证实时价
|
||||
3. 检查是否已在持仓/自选（避免重复）
|
||||
4. 写入 holding_strategies (decision_type='自选策略', status='active')
|
||||
5. 标记 candidates.promoted=1
|
||||
从 candidates 表读未提拔的候选，评估后自动加入 holding_strategies。
|
||||
"""
|
||||
|
||||
import json, os, sqlite3, subprocess, sys, time
|
||||
import sys, json, sqlite3
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
|
||||
BASE = Path("/home/hmo/MoFin")
|
||||
DATA = BASE / "data"
|
||||
DB_PATH = DATA / "mofin.db"
|
||||
|
||||
sys.path.insert(0, str(BASE / "scripts"))
|
||||
from mofin_db import get_conn, write_holding_strategy
|
||||
|
||||
|
||||
def get_quote(code):
|
||||
"""用 stock_quote.py 获取实时行情"""
|
||||
try:
|
||||
r = subprocess.run(
|
||||
[sys.executable, str(BASE / "scripts/stock_quote.py"), str(code)],
|
||||
capture_output=True, text=True, timeout=15
|
||||
)
|
||||
if r.returncode != 0:
|
||||
return None
|
||||
for line in r.stdout.strip().split("\n"):
|
||||
if line.startswith("{"):
|
||||
return json.loads(line)
|
||||
except Exception:
|
||||
return None
|
||||
return None
|
||||
|
||||
|
||||
def already_in_system(conn, code):
|
||||
"""检查是否已在持仓或自选中"""
|
||||
# holdings
|
||||
cur = conn.execute("SELECT COUNT(*) FROM holdings WHERE code=? AND is_active=1", (code,))
|
||||
if cur.fetchone()[0] > 0:
|
||||
return "持仓"
|
||||
# holding_strategies active
|
||||
cur = conn.execute(
|
||||
"SELECT COUNT(*) FROM holding_strategies WHERE code=? AND status='active'",
|
||||
(code,)
|
||||
)
|
||||
if cur.fetchone()[0] > 0:
|
||||
return "自选"
|
||||
return None
|
||||
|
||||
|
||||
def promote_one(conn, cand):
|
||||
"""将一只候选股提拔为自选股（带策略）"""
|
||||
code = cand["code"]
|
||||
name = cand.get("name", "")
|
||||
|
||||
# 1. 检查是否已在系统内
|
||||
exist = already_in_system(conn, code)
|
||||
if exist:
|
||||
mark_promoted(conn, code, f"已在{exist}")
|
||||
return None
|
||||
|
||||
# 2. 获取实时行情
|
||||
quote = get_quote(code)
|
||||
if not quote or not quote.get("price"):
|
||||
return (code, name, f"取价失败，跳过")
|
||||
|
||||
price = quote["price"]
|
||||
change_pct = quote.get("change_pct", 0)
|
||||
|
||||
# 3. 解析 entry_range
|
||||
entry_str = cand.get("entry_range", "")
|
||||
entry_low = 0
|
||||
entry_high = 0
|
||||
if entry_str and "~" in entry_str:
|
||||
try:
|
||||
parts = entry_str.split("~")
|
||||
entry_low = float(parts[0].strip())
|
||||
entry_high = float(parts[1].strip())
|
||||
except (ValueError, IndexError):
|
||||
pass
|
||||
|
||||
# 如果没entry_range，用价格±3%作为默认区间
|
||||
if entry_low <= 0 or entry_high <= 0:
|
||||
entry_low = round(price * 0.97, 2)
|
||||
entry_high = round(price * 1.03, 2)
|
||||
|
||||
stop_loss = cand.get("stop_loss") or round(entry_low * 0.93, 2)
|
||||
target = cand.get("target") or round(entry_high * 1.15, 2)
|
||||
reason = cand.get("reason", "系统挖掘候选")
|
||||
|
||||
# 4. 检测市场（A股/港股）
|
||||
code_str = str(code).strip()
|
||||
currency = "HKD" if (len(code_str) <= 5 or code_str.startswith("0")) and len(code_str) < 6 else "CNY"
|
||||
# 更精确的检测：港股5位
|
||||
if len(code_str) <= 5:
|
||||
currency = "HKD"
|
||||
elif code_str.startswith("0") and len(code_str) <= 5:
|
||||
currency = "HKD"
|
||||
else:
|
||||
currency = "CNY"
|
||||
|
||||
sector = cand.get("sector", "")
|
||||
|
||||
# 5. 写入 holding_strategies
|
||||
strategy_data = {
|
||||
"version": 1,
|
||||
"price": price,
|
||||
"cost": price,
|
||||
"entry_low": entry_low,
|
||||
"entry_high": entry_high,
|
||||
"stop_loss": stop_loss,
|
||||
"take_profit": target,
|
||||
"currency": currency,
|
||||
"strategy_type": "watch",
|
||||
"action": "观望",
|
||||
"timing_signal": "中性",
|
||||
"rr_ratio": round((target - entry_low) / (entry_low - stop_loss), 2) if stop_loss and (entry_low - stop_loss) > 0 else 0,
|
||||
"stock_category": sector or "未分类",
|
||||
"sector_context": sector or "",
|
||||
"status": "active",
|
||||
"source": "candidates",
|
||||
"reason": reason,
|
||||
"type": "自选策略",
|
||||
"decision_type": "自选策略",
|
||||
"time_horizon": "中线",
|
||||
"note": f"候选股自动推广 {datetime.now().strftime('%Y-%m-%d %H:%M')}",
|
||||
}
|
||||
|
||||
ok, msg = write_holding_strategy(conn, code, name, strategy_data)
|
||||
if ok:
|
||||
mark_promoted(conn, code)
|
||||
return (code, name, f"已加自选 {entry_low}~{entry_high} 止损{stop_loss}")
|
||||
else:
|
||||
return (code, name, f"写入失败: {msg}")
|
||||
|
||||
|
||||
def mark_promoted(conn, code, reason=None):
|
||||
"""标记候选为已推广"""
|
||||
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
if reason:
|
||||
conn.execute(
|
||||
"UPDATE candidates SET promoted=1, promoted_at=?, dropped=1, drop_reason=? WHERE code=?",
|
||||
(now, reason, code)
|
||||
)
|
||||
else:
|
||||
conn.execute(
|
||||
"UPDATE candidates SET promoted=1, promoted_at=? WHERE code=?",
|
||||
(now, code)
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
DB_PATH = Path("/home/hmo/MoFin/data/mofin.db")
|
||||
|
||||
def main():
|
||||
start = time.time()
|
||||
today = datetime.now().strftime("%Y-%m-%d %H:%M")
|
||||
|
||||
conn = get_conn()
|
||||
if not conn:
|
||||
print("[SILENT] 无法连接DB", flush=True)
|
||||
return
|
||||
|
||||
try:
|
||||
# 读待推广候选股
|
||||
rows = conn.execute(
|
||||
"SELECT * FROM candidates WHERE promoted=0 AND dropped=0 ORDER BY created_at ASC"
|
||||
).fetchall()
|
||||
|
||||
if not rows:
|
||||
print("[SILENT] 无待推广候选", flush=True)
|
||||
return
|
||||
|
||||
results = []
|
||||
for r in rows:
|
||||
res = promote_one(conn, dict(r))
|
||||
if res:
|
||||
results.append(res)
|
||||
|
||||
conn.commit()
|
||||
elapsed = time.time() - start
|
||||
|
||||
if results:
|
||||
print(f"候选股自动推广 | {today} | {len(results)}只处理 ({elapsed:.0f}s)")
|
||||
for code, name, msg in results:
|
||||
print(f" {code} {name}: {msg}")
|
||||
else:
|
||||
print("[SILENT] 候选推广结束（无结果）", flush=True)
|
||||
|
||||
finally:
|
||||
conn = sqlite3.connect(str(DB_PATH))
|
||||
conn.row_factory = sqlite3.Row
|
||||
|
||||
# 读未提拔候选（按评分降序）
|
||||
rows = conn.execute("""
|
||||
SELECT * FROM candidates
|
||||
WHERE (promoted IS NULL OR promoted = 0)
|
||||
AND (dropped IS NULL OR dropped = 0)
|
||||
AND score >= 6
|
||||
ORDER BY score DESC
|
||||
""").fetchall()
|
||||
|
||||
if not rows:
|
||||
print("[PROMOTE] 无待提拔候选")
|
||||
conn.close()
|
||||
|
||||
return
|
||||
|
||||
promoted = 0
|
||||
for r in rows:
|
||||
code = str(r["code"])
|
||||
name = r["name"] or code
|
||||
price = r["price"] or 0
|
||||
el = r["entry_low"] or 0
|
||||
eh = r["entry_high"] or 0
|
||||
sl = r["stop_loss"] or 0
|
||||
tp = r["take_profit"] or 0
|
||||
score = r["score"] or 0
|
||||
sector = r["sector"] or ""
|
||||
reason = r["reason"] or ""
|
||||
|
||||
# 查是否已在 holding_strategies
|
||||
exists = conn.execute(
|
||||
"SELECT id FROM holding_strategies WHERE code=? AND status='active'",
|
||||
(code,)
|
||||
).fetchone()
|
||||
if exists:
|
||||
# 标记已提拔但不重复加
|
||||
conn.execute("UPDATE candidates SET promoted=1 WHERE code=?", (code,))
|
||||
print(f" ⏭ {code} {name} 已在自选中，标记promoted")
|
||||
continue
|
||||
|
||||
# 构建策略
|
||||
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
timing_signal = "买入" if score >= 7 else "关注"
|
||||
action = f"市场扫描发现({reason})" if reason else "市场扫描发现"
|
||||
|
||||
conn.execute("""
|
||||
INSERT INTO holding_strategies
|
||||
(code, name, price, entry_low, entry_high, stop_loss, take_profit,
|
||||
timing_signal, action, decision_type, strategy_type, status,
|
||||
rr_ratio, stock_category, created_at, updated_at,
|
||||
sector_context, quality_check)
|
||||
VALUES (?,?,?,?,?,?,?,?,?,'自选策略','scan',
|
||||
'active',0,'关注',?,?,'', 'pending')
|
||||
""", (code, name, price, el, eh, sl, tp, timing_signal, action, now, now))
|
||||
|
||||
conn.execute("UPDATE candidates SET promoted=1 WHERE code=?", (code,))
|
||||
promoted += 1
|
||||
print(f" ✅ {code} {name} 评分{score} → 已加入自选({timing_signal})", flush=True)
|
||||
|
||||
conn.commit()
|
||||
print(f"\n[PROMOTE] 本次提拔{promoted}只", flush=True)
|
||||
|
||||
# 推XMPP
|
||||
if promoted > 0:
|
||||
try:
|
||||
import urllib.request
|
||||
msg = f"📈 自动提拔{promoted}只候选入自选"
|
||||
payload = json.dumps({"to": "hmo@yoin.fun", "body": msg, "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)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
conn.close()
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
Reference in New Issue
Block a user