hmo/MoFin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: data-layering ????live_prices????qt.gtimg.cn(5??)

Browse Source
This commit is contained in:
xxm
2026-08-26 19:49:10 +08:00
parent ad5adbee4a
commit e30889158d
5 changed files with 181 additions and 219 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/profile-scripts/candidate_filter.py
+6 -12
View File
@@ -193,19 +193,13 @@ def stage4_capital_flow(code, name):
stocks = (_j.loads(r[0]) or {}).get("stocks") or {}
info = stocks.get(raw)
if not info or not info.get("flow"):
# 2026-08-24 cache miss→腾讯quote兜底(3s快失败):池外候选拿真实内外盘,不靠中性
# 2026-08-26 分层铁律：消费层不直连腾讯quote；改查 live_prices，查不到返回中性
try:
prefix = "sh" if raw.startswith(("6", "9")) else "sz"
import subprocess as _sp
url = f"http://qt.gtimg.cn/q={prefix}{raw}"
r2 = _sp.run(["curl", "-s", url], capture_output=True, timeout=3)
parts = r2.stdout.decode("gbk", errors="ignore").split("~")
if len(parts) > 8 and parts[7] and parts[8]:
outer, inner = int(float(parts[7])), int(float(parts[8]))
if outer > 0 and inner > 0:
ratio = outer / inner
score = 2 if ratio > 1.3 else (1 if ratio > 1.0 else 0)
return score >= 1, score, f"外/内={ratio:.2f}(腾讯兜底)"
_lp = conn.execute(
"SELECT price, change_pct FROM live_prices WHERE code=?", (raw,)
).fetchone()
if _lp and _lp[0]:
return True, 1, f"实时价{float(_lp[0]):.2f} 无资金流(中性)"
except Exception:
pass
return True, 1, "无资金流数据(中性)"
deploy/profile-scripts/mofin_db.py
+8 -23
View File
@@ -1094,33 +1094,18 @@ def query_latest_market(conn: sqlite3.Connection) -> dict:
# ═══════════════════════════════════════════════════════════════════
def fetch_stock_name_tencent(code: str) -> str | None:
"""腾讯 quote 单票查名（数据层专用，2026-08-22 名称治理）。
A股按 6/9→sh、其余→sz；港股5位0/1开头→hk。查到真名返回，查不到返回 None。
"""按 code 查股票名称（数据层专用，2026-08-22 名称治理；2026-08-26 分层铁律：
消费层不直连腾讯API，改读 stocks 表）。查到真名返回，查不到返回 None。
"""
import urllib.request
code = str(code).strip()
if not code:
return None
if len(code) == 5 and code[0] in "01":
sym = "hk" + code
elif code[0] in "69":
sym = "sh" + code
else:
sym = "sz" + code
try:
url = f"http://qt.gtimg.cn/q={sym}"
req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
with urllib.request.urlopen(req, timeout=5) as resp:
text = resp.read().decode("gbk", errors="ignore")
# 格式: v_sh600110="1~诺德股份~600110~..."
if '="' in text and "~" in text:
payload = text.split('="', 1)[1]
parts = payload.split("~")
if len(parts) > 1:
name = parts[1].strip()
if name and name != code:
return name
_db = sqlite3.connect('/home/hmo/MoFin/data/mofin.db', timeout=5)
_row = _db.execute("SELECT name FROM stocks WHERE code=?", (code,)).fetchone()
_db.close()
if _row and _row[0] and str(_row[0]).strip() != code:
return str(_row[0]).strip()
except Exception:
pass
return None
@@ -1893,7 +1878,7 @@ def flush_rec_digest(max_items=5):
weak.append(_d)
if weak:
print(f" [换仓] LLM推荐 {len(weak)} 只", flush=True)
weak = sorted(weak, key=lambda w: ({'弱势持有': 0, '观望': 1}.get(w['timing_signal'], 2),
weak = sorted(weak, key=lambda w: ({'弱势持有': 0, '观望': 1}.get(w['timing_signal'], 2),
-(w['position_pct'] or 0)))
if weak:
need_pct = queued[0][1]
deploy/profile-scripts/per_stock_reassess.py
+32 -27
View File
@@ -111,43 +111,48 @@ def _build_full_analysis(code, entry, result):
elif _mood:
macro_desc += f" 情绪={_mood}"
if not macro_desc:
# fallback: 直接用腾讯API拉大盘
# fallback: 读 stock_daily 大盘指数最近收盘（2026-08-26 分层铁律：消费层不直连腾讯API）
try:
_r2 = __import__('subprocess').run(["curl", "-s", "http://qt.gtimg.cn/q=sh000001,sz399001,sz399006,sh000688"],
capture_output=True, timeout=10)
_txt = _r2.stdout.decode("gbk", errors="ignore")
_idx_defs = [
("sh000001", "上证指数"), ("sz399001", "深证成指"),
("sz399006", "创业板指"), ("sh000688", "科创50"),
]
_parts = []
for _line in _txt.strip().split("\n"):
if "~" not in _line: continue
_p = _line.split("~")
if len(_p) < 4: continue
_name2 = _p[1]
_price2 = _p[3]
_chg2 = _p[32] if len(_p) > 32 else "0"
_parts.append(f"{_name2}({_price2},{_chg2}%)")
for _ic, _iname in _idx_defs:
_rows = _db.execute(
"SELECT date, close FROM stock_daily WHERE code=? ORDER BY date DESC LIMIT 2", (_ic,)
).fetchall()
if not _rows:
continue
_close_i = _rows[0][1]
_chg_i = 0.0
if len(_rows) > 1 and _rows[1][1]:
_chg_i = (_close_i / _rows[1][1] - 1) * 100
_parts.append(f"{_iname}({_close_i:.0f},{_chg_i:+.1f}%)")
if _parts:
macro_desc = "腾讯实时 " + " ".join(_parts[:3])
macro_desc = " ".join(_parts[:4])
except:
pass
# 基本面+实时价：直接从腾讯API拉（盘后也有收盘价）
# 基本面+实时价：读 DB（live_prices + stock_fundamentals，2026-08-26 分层铁律：消费层不直连腾讯API）
try:
_pfx = "sh" if str(code).startswith(("6", "9")) else "sz"
_r3 = __import__('subprocess').run(["curl", "-s", f"http://qt.gtimg.cn/q={_pfx}{code}"],
capture_output=True, timeout=10)
_txt3 = _r3.stdout.decode("gbk", errors="ignore")
_p3 = _txt3.split("~")
if len(_p3) > 45:
_pe = _p3[39] if _p3[39] else ""
_pb = _p3[40] if len(_p3) > 40 and _p3[40] else ""
_mcap = _p3[44] if len(_p3) > 44 and _p3[44] else ""
_price_now = float(_p3[3]) if _p3[3] else 0
_chg_now = float(_p3[32]) if len(_p3) > 32 and _p3[32] else 0
_lp = _db.execute(
"SELECT price, change_pct FROM live_prices WHERE code=?", (str(code),)
).fetchone()
if _lp and _lp[0]:
_price_now = float(_lp[0])
if _price_now > 0:
price = _price_now # 覆盖策略中的price=0
_fs = _db.execute(
"SELECT pe, pb, mcap_total FROM stock_fundamentals WHERE code=? ORDER BY updated_at DESC LIMIT 1", (str(code),)
).fetchone()
if _fs:
_pe = _fs[0]
_pb = _fs[1]
_mcap = _fs[2]
if _pe: pe_val = f"PE={_pe}"
if _pb: pb_val = f"PB={_pb}"
if _mcap:
mcap_val = f"市值{float(_mcap)/10000:.1f}亿" if float(_mcap) > 10000 else f"市值{_mcap}万"
if _mcap:
mcap_val = f"市值{float(_mcap):.1f}亿" # stock_fundamentals.mcap_total 单位=亿元
pe_val += f" {mcap_val}" if pe_val else mcap_val
except:
pass
deploy/profile-scripts/strategy_lifecycle.py
+47 -88
View File
@@ -742,7 +742,7 @@ def load_macro_context():
def batch_fetch_prices(codes):
"""获取实时价格。优先从 DB 读取（price_monitor 每 2 分钟更新），失败才拉腾讯 API。"""
"""获取实时价格。优先从 holdings DB 读取（price_monitor 维护），兜底读 live_prices 表（2026-08-26 分层铁律）。"""
if not codes:
return {}
@@ -777,74 +777,42 @@ def batch_fetch_prices(codes):
except Exception:
pass
# Fallback: 腾讯 API（仅当 DB 无数据时）
batch_size = 15
for batch_start in range(0, len(codes), batch_size):
batch = codes[batch_start:batch_start + batch_size]
symbols = []
code_map = {}
for raw_code in batch:
raw_code = str(raw_code).split('_')[0]
if not raw_code:
continue
if len(raw_code) == 5 and raw_code.isdigit():
prefix = "hk"
elif raw_code.startswith(("6", "5")):
prefix = "sh"
else:
prefix = "sz"
sym = f"{prefix}{raw_code}"
symbols.append(sym)
code_map[sym] = raw_code
if not symbols:
continue
url = f"http://qt.gtimg.cn/q={','.join(symbols)}"
max_retries = 2
for attempt in range(max_retries + 1):
try:
r = urllib.request.urlopen(url, timeout=10)
text = r.read().decode("gbk")
except Exception as e:
if attempt < max_retries:
# Fallback: live_prices 表（price_monitor 2分钟级实时价；2026-08-26 分层铁律：消费层不直连腾讯API）
try:
import sqlite3 as _sq
_db = _sq.connect('/home/hmo/MoFin/data/mofin.db', timeout=5)
_codes = []
_seen = set()
for _rc in codes:
_rc = str(_rc).split('_')[0]
if _rc and _rc not in _seen:
_seen.add(_rc)
_codes.append(_rc)
if _codes:
_ph = ",".join("?" * len(_codes))
_rows = _db.execute(
f"SELECT code, price, change_pct FROM live_prices WHERE code IN ({_ph})", _codes
).fetchall()
for _c, _p, _cg in _rows:
if not _p:
continue
print(f" batch_fetch_prices error: {e}", file=sys.stderr)
continue
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_value = line.split("=", 1)[1].strip().strip('"').strip(";")
fields = raw_value.split("~")
if len(fields) < 35:
continue
orig_code = code_map.get(sym)
if not orig_code:
continue
def f(i):
try:
return float(fields[i]) if fields[i].strip() else 0.0
except:
return 0.0
price_raw = f(3)
# 港股：腾讯 API 返回 HKD 原值，价格比较/止损止盈直接用原值
# （仅市值/总资产汇总时由 mo_models.calc_total_assets 折算 CNY）
all_results[orig_code] = {
"price": price_raw, "close": f(4), "high": f(33), "low": f(34),
"code": orig_code,
}
except Exception:
continue
break # Success - break retry loop
_chg = _cg or 0.0
# 港股 live_prices 价格为 HKD 原值，价格比较/止损止盈直接用原值
# （仅市值/总资产汇总时由 mo_models.calc_total_assets 折算 CNY）
_cl = _p / (1 + _chg / 100) if _chg else _p
all_results[_c] = {
"price": _p, "close": _cl, "high": _p, "low": _p,
"code": _c,
}
_db.close()
except Exception as e:
print(f" batch_fetch_prices live_prices error: {e}", file=sys.stderr)
return all_results
def get_price_tencent(code):
"""获取实时价格。优先 DB（price_monitor 维护），失败才拉腾讯。港股价格存 HKD 原值。"""
"""获取实时价格。优先 DB（holdings），兜底 live_prices 表（price_monitor 维护）。港股价格存 HKD 原值。"""
raw_code = str(code).split('_')[0]
if not raw_code:
return None
@@ -864,33 +832,24 @@ def get_price_tencent(code):
except Exception:
pass
# Fallback: 腾讯 API
# Fallback: live_prices 表（price_monitor 2分钟级实时价；2026-08-26 分层铁律：消费层不直连腾讯API）
try:
from mo_models import is_hk_stock
except ImportError:
is_hk_stock = lambda c: len(str(c).strip()) == 5 and str(c).strip().isdigit()
try:
if is_hk_stock(raw_code):
prefix = "hk"
elif raw_code.startswith("6") or raw_code.startswith("5"):
prefix = "sh"
else:
prefix = "sz"
url = f"http://qt.gtimg.cn/q={prefix}{raw_code}"
r = urllib.request.urlopen(url, timeout=5)
fields = r.read().decode("gbk").split('"')[1].split("~")
def f(i):
try:
return float(fields[i]) if fields[i].strip() else 0.0
except:
return 0.0
price = f(3)
return {
"price": price, "close": f(4), "high": f(33), "low": f(34),
"code": raw_code,
}
import sqlite3 as _sq
_db = _sq.connect('/home/hmo/MoFin/data/mofin.db', timeout=5)
_row = _db.execute("SELECT price, change_pct FROM live_prices WHERE code=?", (raw_code,)).fetchone()
_db.close()
if _row and _row[0]:
_chg = _row[1] or 0.0
# 港股 live_prices 价格为 HKD 原值
_cl = _row[0] / (1 + _chg / 100) if _chg else _row[0]
return {
"price": _row[0], "close": _cl, "high": _row[0], "low": _row[0],
"code": raw_code,
}
print(f" get_price live_prices 无数据 {code}", file=sys.stderr)
return None
except Exception as e:
print(f" get_price error {code}: {e}", file=sys.stderr)
print(f" get_price live_prices error {code}: {e}", file=sys.stderr)
return None
deploy/profile-scripts/technical_analysis.py
+88 -69
View File
@@ -14,7 +14,6 @@
import json
import os
import sys
import urllib.request
from datetime import datetime, date
# 确保本文件所在目录可导入（market_config 与之同目录；本模块会被 MoFin 根脚本跨目录 import）
@@ -75,82 +74,102 @@ def get_quote(code):
except:
pass
# 腾讯API获取全量HLC数据
# 行情数据：读 DB（2026-08-26 分层铁律：消费层不直连腾讯API，改读 live_prices + stock_daily）
raw = str(code).split("_")[0]
prefix = _market_prefix(code)
url = f"http://qt.gtimg.cn/q={prefix}{raw}"
_lp_row = None
_sd_rows = None
_nm_row = None
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)}
import sqlite3 as _sq
_db = _sq.connect('/home/hmo/MoFin/data/mofin.db', timeout=5)
_lp_row = _db.execute("SELECT price, change_pct FROM live_prices WHERE code=?", (raw,)).fetchone()
_sd_rows = _db.execute(
"SELECT date, open, close, high, low, volume, amount FROM stock_daily "
"WHERE code=? ORDER BY date DESC LIMIT 2", (raw,)
).fetchall()
_nm_row = _db.execute("SELECT name FROM stocks WHERE code=?", (raw,)).fetchone()
_db.close()
except Exception:
pass
def get(i):
try:
return float(fields[i]) if fields[i].strip() else None
except (IndexError, ValueError):
return None
_lp_price = _lp_row[0] if _lp_row and _lp_row[0] else None
_lp_chg = _lp_row[1] if _lp_row and _lp_row[1] is not None else 0.0
if _lp_price or db_price:
today_str = date.today().isoformat()
price_v = _lp_price or db_price
change_v = _lp_chg if _lp_price else (db_chg or 0)
close_yest_v = price_v / (1 + change_v / 100) if change_v else price_v
# stock_daily 最近一根日K（收盘级 HLC，缺实时盘中H高）
_sd = _sd_rows[0] if _sd_rows else None
_sd_open = _sd[1] if _sd else None
_sd_high = _sd[3] if _sd else None
_sd_low = _sd[4] if _sd else None
_sd_vol = _sd[5] if _sd else None
_sd_amt = _sd[6] if _sd else None
_name_v = _nm_row[0] if _nm_row and _nm_row[0] else code
q = {
"code": raw,
"market": prefix,
"name": _name_v,
"price": price_v,
"close_yest": close_yest_v,
"open": _sd_open if _sd_open else price_v,
"high": _sd_high if _sd_high else price_v,
"low": _sd_low if _sd_low else price_v,
"volume": _sd_vol,
"amount": _sd_amt,
"change": price_v - close_yest_v,
"change_pct": change_v,
"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": "",
"_date": today_str,
}
today_str = date.today().isoformat()
q = {
"code": raw,
"market": prefix,
"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,
}
# 写入价格历史缓存（每日一次；仅最近K线日期==今天才更新，避免用旧日数据冒充今日）
h = _sd_high if _sd_high else price_v
l = _sd_low if _sd_low else price_v
c = price_v
v = _sd_vol
amt = _sd_amt
_sd_date = _sd[0] if _sd else None
if h and l and c and (_sd_date == today_str or _sd is None):
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)
# 写入价格历史缓存（每日一次）
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}}
# 写入60秒缓存
get_quote.__dict__["_cache"] = {**get_quote.__dict__.get("_cache", {}), code: {"ts": now, "data": q}}
return q
return q
if db_price:
return {"code": code, "price": db_price, "change_pct": db_chg or 0}
return {"code": code, "error": "数据不足(无本地行情)"}
def calc_support_resistance(q):