From bd1ef0c9a4573d4e438e37913e237783bc3ce3e1 Mon Sep 17 00:00:00 2001 From: xxm Date: Wed, 26 Aug 2026 19:49:10 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20data-layering=20=E6=B6=88=E8=B4=B9?= =?UTF-8?q?=E5=B1=82=E8=AF=BBlive=5Fprices=E6=9B=BF=E6=8D=A2=E7=9B=B4?= =?UTF-8?q?=E8=BF=9Eqt.gtimg.cn(5=E8=84=9A=E6=9C=AC)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- deploy/profile-scripts/candidate_filter.py | 18 +-- deploy/profile-scripts/mofin_db.py | 31 +--- deploy/profile-scripts/per_stock_reassess.py | 59 +++---- deploy/profile-scripts/strategy_lifecycle.py | 135 ++++++---------- deploy/profile-scripts/technical_analysis.py | 157 +++++++++++-------- 5 files changed, 181 insertions(+), 219 deletions(-) diff --git a/deploy/profile-scripts/candidate_filter.py b/deploy/profile-scripts/candidate_filter.py index 30cf135c..a75511e1 100644 --- a/deploy/profile-scripts/candidate_filter.py +++ b/deploy/profile-scripts/candidate_filter.py @@ -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, "无资金流数据(中性)" diff --git a/deploy/profile-scripts/mofin_db.py b/deploy/profile-scripts/mofin_db.py index 09e6ddf0..b6f851c4 100644 --- a/deploy/profile-scripts/mofin_db.py +++ b/deploy/profile-scripts/mofin_db.py @@ -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] diff --git a/deploy/profile-scripts/per_stock_reassess.py b/deploy/profile-scripts/per_stock_reassess.py index 477504f9..ef2f3c6f 100644 --- a/deploy/profile-scripts/per_stock_reassess.py +++ b/deploy/profile-scripts/per_stock_reassess.py @@ -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 diff --git a/deploy/profile-scripts/strategy_lifecycle.py b/deploy/profile-scripts/strategy_lifecycle.py index c5d88353..ac56e572 100644 --- a/deploy/profile-scripts/strategy_lifecycle.py +++ b/deploy/profile-scripts/strategy_lifecycle.py @@ -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 diff --git a/deploy/profile-scripts/technical_analysis.py b/deploy/profile-scripts/technical_analysis.py index d6489afb..9694cc76 100644 --- a/deploy/profile-scripts/technical_analysis.py +++ b/deploy/profile-scripts/technical_analysis.py @@ -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):