diff --git a/deploy/profile-scripts/accumulation_scanner.py b/deploy/profile-scripts/accumulation_scanner.py index f2bf6def..0ac20100 100644 --- a/deploy/profile-scripts/accumulation_scanner.py +++ b/deploy/profile-scripts/accumulation_scanner.py @@ -13,7 +13,7 @@ 数据源:腾讯批量行情API(日K线+实时价) """ -import sys, json, urllib.request, re, time, os +import sys, json, os from pathlib import Path from datetime import datetime, timedelta from collections import defaultdict @@ -29,48 +29,71 @@ DB_PATH = Path("/home/hmo/MoFin/data/mofin.db") UA = "Mozilla/5.0" def fetch_qq_batch(symbols): - """腾讯批量实时行情""" + """批量行情:读 DB live_prices + stock_daily(2026-08-26 分层铁律:消费层不直连腾讯API)""" if not symbols: return {} + import sqlite3 results = {} - # 分批,每批100个(腾讯推荐上限) - for i in range(0, len(symbols), 100): - batch = symbols[i:i+100] - url = f"http://qt.gtimg.cn/q={','.join(batch)}" - 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") - 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 - prev_close = float(parts[4]) if parts[4] else 0 - high = float(parts[33]) if parts[33] else 0 - low = float(parts[34]) if parts[34] else 0 - volume = int(float(parts[6])) if parts[6] else 0 # 股数 - amount = float(parts[37]) if len(parts) > 37 and parts[37] else 0 - change_pct = float(parts[32]) if parts[32] else 0 - # 市盈率 - pe = float(parts[39]) if len(parts) > 39 and parts[39] else 0 - # 流通市值 - mcap = float(parts[44]) if len(parts) > 44 and parts[44] else 0 - if price > 0 and volume > 0: - results[code] = { - "code": code, "name": name, "price": price, - "prev_close": prev_close, "high": high, "low": low, - "volume": volume, "amount": amount, - "change_pct": change_pct, "pe": pe, "mcap": mcap, - } - except Exception as e: - print(f" 批量查询错误: {e}", file=sys.stderr) - time.sleep(0.15) # 批次间隔150ms(腾讯建议100ms以上,留余量) + # 提取纯代码(去掉 sh/sz/hk 前缀,港股5位如 00700) + codes = [] + for s in symbols: + c = str(s).strip() + for pfx in ("sh", "sz", "hk"): + if c.startswith(pfx): + c = c[len(pfx):] + break + if c and c not in codes: + codes.append(c) + if not codes: + return results + try: + conn = sqlite3.connect(str(DB_PATH), timeout=5) + conn.execute("PRAGMA busy_timeout=30000") + # 实时价:一次查 + ph = ",".join("?" * len(codes)) + price_rows = conn.execute( + f"SELECT code, price, change_pct FROM live_prices WHERE code IN ({ph})", codes + ).fetchall() + prices = {r[0]: (r[1], r[2]) for r in price_rows} + # 名称 + name_rows = conn.execute( + f"SELECT code, name FROM stocks WHERE code IN ({ph})", codes + ).fetchall() + names = {r[0]: r[1] or "" for r in name_rows} + # 最近日K(昨收/高低/量/额) + sd = {} + for c in codes: + row = conn.execute( + "SELECT close, high, low, volume, amount FROM stock_daily " + "WHERE code=? ORDER BY date DESC LIMIT 1", (c,) + ).fetchone() + if row: + sd[c] = row + conn.close() + + for code in codes: + if code not in prices: + continue + price = prices[code][0] + change_pct = prices[code][1] or 0 + row = sd.get(code) + if not price or price <= 0: + continue + if not row or not row[3]: # 无日K或无量 → 跳过(中性) + continue + prev_close = row[0] or 0 + high = row[1] or 0 + low = row[2] or 0 + volume = int(row[3]) if row[3] else 0 # 股数 + amount = row[4] or 0 + # DB 无 PE/流通市值 → 中性 0(detect 中 pe/mcap 条件不贡献分) + results[code] = { + "code": code, "name": names.get(code, code), "price": price, + "prev_close": prev_close, "high": high, "low": low, + "volume": volume, "amount": amount, + "change_pct": change_pct, "pe": 0, "mcap": 0, + } + except Exception as e: + print(f" 批量查询错误: {e}", file=sys.stderr) return results def get_stock_pool(): diff --git a/deploy/profile-scripts/chip_factors.py b/deploy/profile-scripts/chip_factors.py index 6ee95c0e..31b8f9d5 100644 --- a/deploy/profile-scripts/chip_factors.py +++ b/deploy/profile-scripts/chip_factors.py @@ -16,7 +16,7 @@ results = cf.batch_calc(["600519", "00700", "000700"]) """ -import json, os, sqlite3, time, urllib.request +import json, sqlite3, time from datetime import datetime, timedelta from mo_data import read_decisions, get_price from pathlib import Path @@ -25,8 +25,6 @@ DB_PATH = Path("/home/hmo/MoFin/data/mofin.db") MOFIN_ROOT = Path("/home/hmo/MoFin") CACHE_DIR = MOFIN_ROOT / "data" / "chip_cache" -# ── 分钟数据限流 ── -_last_minute_call = 0 def _fetch_quote(code): """拉实时价,统一走 mo_data.get_price""" @@ -38,21 +36,17 @@ def _fetch_quote(code): def _fetch_minute_kline(code, count=60): - """拉1分钟K线(带限流)""" - global _last_minute_call - now = time.time() - if now - _last_minute_call < 1.0: - time.sleep(1.0 - (now - _last_minute_call)) - secid = f"1.{code}" if code.startswith(('6','5')) else f"0.{code}" - url = (f"https://push2.eastmoney.com/api/qt/stock/kline/get" - f"?secid={secid}&fields1=f1,f2,f3&fields2=f51,f52,f53,f54,f55,f56,f57" - f"&klt=1&fqt=1&end=20500101&lmt={min(count, 240)}") + """拉日K(读 stock_daily,2026-08-26 分层铁律:消费层不直连东财push2)""" try: - req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) - resp = urllib.request.urlopen(req, timeout=8) - data = json.loads(resp.read())["data"]["klines"] - _last_minute_call = time.time() - return [line.split(",") for line in data] + import sqlite3 + conn = sqlite3.connect(str(DB_PATH), timeout=5) + rows = conn.execute( + "SELECT date, open, close, high, low, volume FROM stock_daily " + "WHERE code=? ORDER BY date DESC LIMIT ?", (str(code), min(count, 240)) + ).fetchall() + conn.close() + # 结构对齐原分钟K:m[5] = volume + return [[str(r[0]), r[1], r[2], r[3], r[4], r[5] or 0] for r in rows] except: return None @@ -85,23 +79,21 @@ class ChipFactors: # ── 筹码分布估算(用日线OHLCV) ── def _build_chip_distribution(self, code): """从日线K线估算筹码分布。 - + 原理:假设每日成交量在OHLC区间内均匀分布, 每根K线的成交量按价格区间分配,累积成筹码分布。 + (2026-08-26 分层铁律:读 DB stock_daily,不直连腾讯API) """ - for k in list(os.environ.keys()): - if 'proxy' in k.lower(): - os.environ.pop(k) - # 从腾讯API取60日K线 - prefix = "sh" if code.startswith(('60','68','51')) else "sz" if code.startswith(('00','30','15')) else "hk" - url = f"http://ifzq.gtimg.cn/appstock/app/fqkline/get?param={prefix}{code},day,,,640,qfq" try: - opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) - req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) - resp = opener.open(req, timeout=8).read().decode('utf-8') - data = json.loads(resp) - day_key = 'qfqday' if prefix != 'hk' else 'day' - bars = data.get('data', {}).get(f'{prefix}{code}', {}).get(day_key, []) + import sqlite3 + conn = sqlite3.connect(str(DB_PATH), timeout=5) + rows = conn.execute( + "SELECT date, open, close, high, low, volume FROM stock_daily " + "WHERE code=? ORDER BY date DESC LIMIT 640", (str(code),) + ).fetchall() + conn.close() + # 对齐原 qfqday bar:[date, open, close, high, low, volume] + bars = [list(r) for r in rows] except: return {} @@ -169,16 +161,17 @@ class ChipFactors: prev_winner = prev.get("winner_pct", winner_pct) prev_bias = prev.get("bias", 0) - # 估算换手率(近10日均量/总流通股) + # 估算换手率(近10日均量/总流通股;2026-08-26 分层铁律:读 stock_daily,不直连腾讯API) turnover = 0.02 # 默认2% try: - prefix2 = "sh" if code.startswith(('60','68','51','56','50')) else "sz" if code.startswith(('00','30','15')) else "hk" - url2 = f"http://ifzq.gtimg.cn/appstock/app/fkline/get?param={prefix2}{code},day,,,10,qfq" - req2 = urllib.request.Request(url2, headers={"User-Agent": "Mozilla/5.0"}) - resp2 = urllib.request.urlopen(req2, timeout=5).read().decode('utf-8') - data2 = json.loads(resp2) - dk = 'qfqday' if prefix2 != 'hk' else 'day' - bars2 = data2.get('data', {}).get(f'{prefix2}{code}', {}).get(dk, []) + import sqlite3 + conn2 = sqlite3.connect(str(DB_PATH), timeout=5) + rows2 = conn2.execute( + "SELECT date, open, close, high, low, volume FROM stock_daily " + "WHERE code=? ORDER BY date DESC LIMIT 10", (str(code),) + ).fetchall() + conn2.close() + bars2 = [list(r) for r in rows2] if len(bars2) > 5: avg_vol = sum(float(b[5]) for b in bars2[-10:] if len(b)>5) / min(len(bars2), 10) # 用近60日最高量估算总流通股 diff --git a/deploy/profile-scripts/collect_evaluation_data.py b/deploy/profile-scripts/collect_evaluation_data.py index 687eed19..325c9fa7 100644 --- a/deploy/profile-scripts/collect_evaluation_data.py +++ b/deploy/profile-scripts/collect_evaluation_data.py @@ -16,10 +16,8 @@ D6 资金面 — 成交额/换手率/量比 """ import json -import urllib.request import os import sys -import re from datetime import datetime from pathlib import Path @@ -34,8 +32,6 @@ DATA_DIR = Path(__file__).parent.parent / "data" PROFILES_PATH = DATA_DIR / "stock_profiles.json" OUTPUT_PATH = DATA_DIR / "evaluation_input.json" -UA = "Mozilla/5.0" - def load_json(path, default=None): try: @@ -52,52 +48,68 @@ def save_json(path, data): def fetch_tencent_data(symbols): - """批量拉行情。DB 优先,腾讯 API fallback""" + """批量拉行情。读 DB live_prices + stock_daily(2026-08-26 分层铁律:消费层不直连腾讯API)""" if not symbols: return {} - # DB 优先 + import sqlite3 + result = {} try: - from mofin_db import get_prices_batch_from_db - db = get_prices_batch_from_db(symbols) - if db: - return {code: {"name": "", "price": p, "prev_close": 0, "change_pct": chg or 0, - "high": 0, "low": 0} for code, (p, chg) in db.items()} - except: pass - # Fallback: 腾讯 - code_map = {} - query_symbols = [] - for c in symbols: - sym = f"hk{c}" if len(c) == 5 else f"sh{c}" if c.startswith(("5", "6", "9")) else f"sz{c}" - query_symbols.append(sym) - code_map[sym] = c - url = f"http://qt.gtimg.cn/q={','.join(query_symbols)}" - try: - req = urllib.request.Request(url, headers={"User-Agent": UA}) - resp = urllib.request.urlopen(req, timeout=15) - text = resp.read().decode("gbk") + conn = sqlite3.connect('/home/hmo/MoFin/data/mofin.db', timeout=5) + syms = list(symbols) + ph = ",".join("?" * len(syms)) + # 实时价:一次查 + price_rows = conn.execute( + f"SELECT code, price, change_pct FROM live_prices WHERE code IN ({ph})", syms + ).fetchall() + prices = {r[0]: (r[1], r[2]) for r in price_rows} + # 名称 + name_rows = conn.execute( + f"SELECT code, name FROM stocks WHERE code IN ({ph})", syms + ).fetchall() + names = {r[0]: r[1] or "" for r in name_rows} + # 最近日K(昨收/今开/高低/量) + sd = {} + for c in syms: + row = conn.execute( + "SELECT close, open, high, low, volume FROM stock_daily " + "WHERE code=? ORDER BY date DESC LIMIT 1", (c,) + ).fetchone() + if row: + sd[c] = row + conn.close() + + for code in syms: + if code not in prices: + continue + price = prices[code][0] + change_pct = prices[code][1] or 0 + if not price or price <= 0: + continue + row = sd.get(code) + row_c = row[0] if row else 0 + row_o = row[1] if row else 0 + row_h = row[2] if row else 0 + row_l = row[3] if row else 0 + row_v = row[4] if row else 0 + result[code] = { + "name": names.get(code, ""), + "price": price, + "prev_close": row_c, + "open": row_o, + "change_pct": change_pct, + "high": row_h, + "low": row_l, + "volume": row_v, + } except Exception as e: print(f"行情拉取失败: {e}", file=sys.stderr) return {} - result = {} - for line in text.strip().split("\n"): - line = line.strip() - if not line or "=" not in line: - continue - raw = line.split("=", 1)[1].strip().strip('"').strip(";") - fields = raw.split("~") - if len(fields) < 35: - continue - sym = line.split("=", 1)[0].strip().lstrip("v_") - orig = code_map.get(sym) - if not orig: - continue - # 统一格式(A股和港股字段长度不同) - result[orig] = fields return result def fetch_indices(): - """拉五大指数""" + """拉指数:读 DB stock_daily 最近收盘(2026-08-26 分层铁律:消费层不直连腾讯API)""" + import sqlite3 index_codes = { "sh000001": "上证指数", "sz399001": "深证成指", @@ -105,37 +117,33 @@ def fetch_indices(): "hkHSI": "恒生指数", "hkHSTECH": "恒生科技", } - idx_map = {} - for c, n in index_codes.items(): - sym = c # 已经是完整符号 - idx_map[sym] = n - url = f"http://qt.gtimg.cn/q={','.join(index_codes.keys())}" + result = {} try: - req = urllib.request.Request(url, headers={"User-Agent": UA}) - resp = urllib.request.urlopen(req, timeout=10) - text = resp.read().decode("gbk") + conn = sqlite3.connect('/home/hmo/MoFin/data/mofin.db', timeout=5) + for c, n in index_codes.items(): + rows = conn.execute( + "SELECT date, close, open, high, low FROM stock_daily " + "WHERE code=? ORDER BY date DESC LIMIT 2", (c,) + ).fetchall() + if not rows: + continue # 读不到则跳过(中性) + latest = rows[0] + prev = rows[1] if len(rows) > 1 else None + prev_close = prev[1] if prev else latest[1] + price = latest[1] or 0 + change_pct = (price - prev_close) / prev_close * 100 if prev_close else 0 + result[n] = { + "price": safe_float(price), + "prev_close": safe_float(prev_close if prev else None), + "change_pct": safe_float(change_pct), + "high": safe_float(latest[3] or price), + "low": safe_float(latest[4] or price), + "timestamp": latest[0], + } + conn.close() except Exception as e: print(f"指数拉取失败: {e}", file=sys.stderr) return {} - result = {} - for line in text.strip().split("\n"): - line = line.strip() - if not line or "=" not in line: - continue - raw = line.split("=", 1)[1].strip().strip('"').strip(";") - fields = raw.split("~") - if len(fields) < 33: - continue - sym = line.split("=", 1)[0].strip().lstrip("v_") - name = idx_map.get(sym, sym) - result[name] = { - "price": safe_float(fields[3]), - "prev_close": safe_float(fields[4]), - "change_pct": safe_float(fields[32]), - "high": safe_float(fields[33]), - "low": safe_float(fields[34]), - "timestamp": fields[30] if len(fields) > 30 else "", - } return result diff --git a/deploy/profile-scripts/divergence_detector.py b/deploy/profile-scripts/divergence_detector.py index 30ca6c81..a07f7f18 100644 --- a/deploy/profile-scripts/divergence_detector.py +++ b/deploy/profile-scripts/divergence_detector.py @@ -14,7 +14,7 @@ divergence_detector.py — 跨市场背离监测器(no_agent) - 状态文件 macro_divergence_state.json - no_agent: 有信号才出声 """ -import sys, json, re, datetime, os, requests +import sys, json, datetime, os from pathlib import Path # ── 消息通道统一路由(broadcast/xmpp by delivery) ── @@ -47,69 +47,45 @@ DIVERGENCE_MODERATE = 3.0 # >3% → moderate信号 STREAK_DAYS = 3 # 连涨/连跌3天 → 信号 def fetch_indices(): - """获取所有指数实时数据(直接调用新浪API,mo_data不支持指数代码)""" - import requests - sina_map = { - "sh000001": "s_sh000001", - "sz399001": "s_sz399001", - "sz399006": "s_sz399006", - "sh000688": "s_sh000688", - "sh000016": "s_sh000016", - "sh000300": "s_sh000300", - "hkHSI": "rt_hkHSI", - "hkHSCEI": "rt_hkHSCEI", + """获取所有指数数据(读 DB stock_daily 收盘价,2026-08-26 分层铁律:消费层不直连新浪API)""" + import sqlite3 + # 指数代码 → 中文名(stock_daily 表已有 sh000001/sz399001/sz399006/sh000688) + index_map = { + "sh000001": "上证指数", + "sz399001": "深证成指", + "sz399006": "创业板指", + "sh000688": "科创50", + "sh000016": "上证50", + "sh000300": "沪深300", } - url = "http://hq.sinajs.cn/list=" + ",".join(sina_map.values()) - headers = {"Referer": "https://finance.sina.com.cn"} + indices = {} try: - r = requests.get(url, headers=headers, timeout=10) - r.encoding = "gbk" + conn = sqlite3.connect(str(DB_PATH), timeout=5) + for sym, cname in index_map.items(): + rows = conn.execute( + "SELECT date, close, high, low FROM stock_daily " + "WHERE code=? ORDER BY date DESC LIMIT 2", (sym,) + ).fetchall() + if not rows: + continue + latest = rows[0] + prev_close = rows[1][1] if len(rows) > 1 else latest[1] + price = latest[1] or 0 + change_pct = (price - prev_close) / prev_close * 100 if prev_close else 0 + indices[sym] = { + "name": cname, + "price": price, + "close": price, + "change_pct": round(change_pct, 2), + "high": latest[2] or 0, + "low": latest[3] or 0, + "timestamp": latest[0], + } + conn.close() except Exception as e: print(f"[DIVERGE] 采集失败: {e}", file=sys.stderr) return {} - - indices = {} - for line in r.text.strip().split("\n"): - line = line.strip() - if not line: - continue - # Parse: var hq_str_XXXX="fields,..."; - try: - var_name = line.split('"')[0].rsplit("_", 1)[-1].rstrip("=") - fields = line.split('"')[1].split(",") - except (IndexError, ValueError): - continue - - # Find the symbol - sym = None - for s, sn in sina_map.items(): - if sn.endswith(var_name): - sym = s - break - if not sym: - continue - - if sym.startswith("hk"): - # HK: name,price,open,high,low,prev_close,change,change_pct,... - name = fields[1] - price = float(fields[2]) if fields[2] else 0 - change_pct = float(fields[8]) if len(fields) > 8 and fields[8] else 0 - else: - # A-share: name,price,change,change_pct,... - name = fields[0] - price = float(fields[1]) if fields[1] else 0 - change_pct = float(fields[3]) if len(fields) > 3 and fields[3] else 0 - - indices[sym] = { - "name": name, - "price": price, - "close": 0, - "change_pct": change_pct, - "high": 0, - "low": 0, - "timestamp": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"), - } - + # 港股指数 hkHSI/hkHSCEI 不在 stock_daily,读不到则不返回(中性值,由调用方跳过低检测) return indices def load_history(): diff --git a/deploy/profile-scripts/mo_provider.py b/deploy/profile-scripts/mo_provider.py index 142e2505..4ba5f25b 100644 --- a/deploy/profile-scripts/mo_provider.py +++ b/deploy/profile-scripts/mo_provider.py @@ -181,53 +181,38 @@ class MoDataProvider: return None - # ── 分钟级 K 线 ────────────────────────────────────────────── - - _last_minute_call = 0 # 限流时间戳 + # ── K 线(日级,读 DB stock_daily)──────────────────────────── def get_minute_kline(self, code: str, count: int = 60) -> list | None: - """获取1分钟K线数据(东方财富 push2)。 - - 限流保护:每次调用间隔至少1秒,批量查询间隔2秒。 + """获取K线数据(读 DB stock_daily,2026-08-26 分层铁律:消费层不直连东财push2)。 Args: - code: 股票代码(6位,如'600519') - count: 获取条数(最大240,约4小时) + code: 股票代码(6位,如'600519';港股5位如'00700') + count: 获取条数(最大240) Returns: - [{"time":"09:31","open":xx,"close":xx,"high":xx,"low":xx,"volume":xx,"amount":xx}, ...] + [{"time":"YYYY-MM-DD","open":xx,"close":xx,"high":xx,"low":xx,"volume":xx,"amount":xx}, ...] 或 None """ - import time, urllib.request - now = time.time() - elapsed = now - self._last_minute_call - if elapsed < 1.0: - time.sleep(1.0 - elapsed) - - # A股secid: 1.上海 0.深圳 - secid = f"1.{code}" if code.startswith(('6','5')) else f"0.{code}" - url = (f"https://push2.eastmoney.com/api/qt/stock/kline/get" - f"?secid={secid}&fields1=f1,f2,f3&fields2=f51,f52,f53,f54,f55,f56,f57" - f"&klt=1&fqt=1&end=20500101&lmt={min(count, 240)}") - try: - req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) - resp = urllib.request.urlopen(req, timeout=8) - data = json.loads(resp.read())["data"]["klines"] + import sqlite3 + conn = sqlite3.connect('/home/hmo/MoFin/data/mofin.db', timeout=5) + rows = conn.execute( + "SELECT date, open, close, high, low, volume, amount FROM stock_daily " + "WHERE code=? ORDER BY date DESC LIMIT ?", (str(code), min(count, 240)) + ).fetchall() + conn.close() result = [] - for line in data: - parts = line.split(",") - if len(parts) >= 6: - result.append({ - "time": parts[0][-5:], # "2026-07-01 09:31" → "09:31" - "open": float(parts[1]), - "close": float(parts[2]), - "high": float(parts[3]), - "low": float(parts[4]), - "volume": int(parts[5]), - "amount": float(parts[6]) if len(parts) > 6 else 0, - }) - self._last_minute_call = time.time() + for r in rows: + result.append({ + "time": str(r[0]), + "open": float(r[1]), + "close": float(r[2]), + "high": float(r[3]), + "low": float(r[4]), + "volume": int(r[5] or 0), + "amount": float(r[6] or 0), + }) return result except Exception as e: logger.warning("get_minute_kline(%s) 失败: %s", code, e) diff --git a/deploy/profile-scripts/multi_timeframe.py b/deploy/profile-scripts/multi_timeframe.py index b3607f93..f199005a 100644 --- a/deploy/profile-scripts/multi_timeframe.py +++ b/deploy/profile-scripts/multi_timeframe.py @@ -12,8 +12,6 @@ import json import os -import urllib.request -import urllib.error from datetime import datetime, date, timedelta from typing import Optional @@ -23,11 +21,7 @@ DATA_DIR = "/home/hmo/web-dashboard/data" HISTORY_PATH = os.path.join(DATA_DIR, "price_history.json") # multi_tf_cache.json 已迁移到 DB (mtf_cache 表) -# 腾讯API K线端点 -KLINE_URL = "http://web.ifzq.gtimg.cn/appstock/app/fqkline/get?param={market}{code},{period},,,{count},qfq" - -# 腾讯实时行情端点(用于市场前缀判断) -QUOTE_URL = "http://qt.gtimg.cn/q={market}{code}" +# (2026-08-26 分层铁律:K线/行情读 DB stock_daily/live_prices,不再直连腾讯API) def _write_klines_to_db(code: str, daily: list, weekly: list, monthly: list, fundamentals: dict = None): @@ -125,8 +119,42 @@ def _save_mtf_cache(): pass +def _aggregate_bars(daily_bars: list, period: str) -> list: + """把升序日K聚合为周K/月K(period: week/month)。 + + Args: + daily_bars: 升序日K列表 [{date,open,close,high,low,volume}] + period: "week" / "month" + + Returns: + 升序聚合K线 [{date,open,close,high,low,volume}] + """ + from collections import OrderedDict + groups = OrderedDict() + for b in daily_bars: + try: + dt = datetime.strptime(b["date"], "%Y-%m-%d") + except Exception: + continue + if period == "week": + iso = dt.isocalendar() + key = (iso[0], iso[1]) # (年, ISO周) + else: # month + key = (dt.year, dt.month) + if key not in groups: + groups[key] = dict(b) + else: + g = groups[key] + g["close"] = b["close"] + g["high"] = max(g["high"], b["high"]) + g["low"] = min(g["low"], b["low"]) + g["volume"] += b["volume"] + g["date"] = b["date"] + return list(groups.values()) + + def fetch_kline(code: str, period: str = "day", count: int = 120) -> list: - """从腾讯API获取K线数据,优先使用本地缓存 + """从 DB stock_daily 获取K线数据(2026-08-26 分层铁律:消费层不直连腾讯API) Args: code: 股票代码 (如 "300548") @@ -151,61 +179,47 @@ def fetch_kline(code: str, period: str = "day", count: int = 120) -> list: if cached_klines and updated_at and (now - updated_at) < _KLINE_CACHE_TTL.get(period, 3600): return cached_klines - market = _market_prefix(code) - is_index = any(code.startswith(p) for p in ["sh", "sz", "hk"]) - - # 指数代码已经自带前缀,API直接用code;普通股票需要加market前缀 - api_code = code if is_index else f"{market}{code}" - url = f"http://web.ifzq.gtimg.cn/appstock/app/fqkline/get?param={api_code},{period},,,{count},qfq" + # 读 DB stock_daily(2026-08-26 分层铁律:消费层不直连腾讯API) + # 指数代码(sh000001/sz399001)与港股(5位 00700)在 stock_daily 中直接以原 code 存储 + raw_code = str(code).split("_")[0] + + # 周/月K需要更多日K做聚合:周K≈5日/根,月K≈22日/根,留余量 + need = count + if period in ("week", "month"): + need = count * (5 if period == "week" else 22) + 10 try: - req = urllib.request.Request(url, headers=_user_agent()) - with urllib.request.urlopen(req, timeout=10) as resp: - raw = json.loads(resp.read().decode("utf-8")) + import sqlite3 + db = sqlite3.connect('/home/hmo/MoFin/data/mofin.db', timeout=5) + rows = db.execute( + "SELECT date, open, close, high, low, volume FROM stock_daily " + "WHERE code=? ORDER BY date DESC LIMIT ?", (raw_code, need) + ).fetchall() + db.close() except Exception as e: return {"error": str(e), "code": code, "period": period} - if not isinstance(raw, dict): - return {"error": f"API returned {type(raw).__name__}", "raw": str(raw)[:200]} - - api_data = raw.get("data", {}) - if not isinstance(api_data, dict): - return {"error": f"data field is {type(api_data).__name__}", "raw": str(api_data)[:200]} - - # 指数代码已经自带前缀(sh000001/sz399001),直接用 - # 普通股票代码需要加market前缀(sh600036/sz300750) - is_index = any(code.startswith(p) for p in ["sh", "sz", "hk"]) - stock_key = code if is_index else f"{market}{code}" - stock_data = api_data.get(stock_key, {}) - - # 腾讯API的K线字段名: qfqday, qfqweek, qfqmonth - period_key = f"qfq{period}" - klines = stock_data.get(period_key, []) - - if not klines: - # 尝试其他字段名 - for k in stock_data: - if isinstance(stock_data[k], list) and len(stock_data[k]) > 0: - if isinstance(stock_data[k][0], list) and len(stock_data[k][0]) >= 6: - klines = stock_data[k] - break - - result = [] - for k in klines: - if len(k) >= 6: - try: - result.append({ - "date": str(k[0]), - "open": float(k[1]), - "close": float(k[2]), - "high": float(k[3]), - "low": float(k[4]), - "volume": float(k[5]), - }) - except (ValueError, IndexError): + bars = [] + for r in reversed(rows): # 升序(旧→新),与腾讯API输出时序一致 + try: + if r[1] is None or r[2] is None: continue + bars.append({ + "date": str(r[0]), + "open": float(r[1]), + "close": float(r[2]), + "high": float(r[3] or r[2]), + "low": float(r[4] or r[2]), + "volume": float(r[5] or 0), + }) + except (ValueError, TypeError): + continue - return result + if period == "day": + return bars[-count:] + + # 周/月K:由日K聚合(open=首日open, close=末日close, high=max, low=min, volume=sum) + return _aggregate_bars(bars, period)[-count:] def calc_moving_averages(klines: list, windows: list = [5, 10, 20, 60]) -> dict: diff --git a/deploy/profile-scripts/strategy-staleness-check.py b/deploy/profile-scripts/strategy-staleness-check.py index 274551f5..8b5c5031 100644 --- a/deploy/profile-scripts/strategy-staleness-check.py +++ b/deploy/profile-scripts/strategy-staleness-check.py @@ -10,7 +10,7 @@ 输出两份:JSON报告(给系统) + 人类可读摘要(stdout for cron) """ -import json, sys, os, re, urllib.request, sqlite3 +import json, sys, os, re, sqlite3 sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..')) from datetime import datetime from mo_data import read_decisions, read_portfolio @@ -31,22 +31,20 @@ DIVERGENCE_WARN = 30 # 偏离买入区>30%→警告 DIVERGENCE_CRIT = 50 # 偏离>50%→严重 def get_price(code): - """从腾讯API获取当前价""" + """从 live_prices 表获取当前价(2026-08-26 分层铁律:消费层不直连腾讯API)""" try: - market = "sh" if code.startswith("6") else "sz" if code.startswith("0") or code.startswith("3") else "" - if code.startswith(("00", "30")) or code.startswith("68"): - market = "sh" if code.startswith("6") else "sz" - elif code.startswith(("01", "02", "03")): - market = "sz" - url = f"http://qt.gtimg.cn/q={market}{code}" - req = urllib.request.Request(url, headers={"User-Agent": "curl/7.81"}) - with urllib.request.urlopen(req, timeout=5) as resp: - raw = resp.read().decode("gbk") - parts = raw.split("~") - if len(parts) > 3: - price = float(parts[3]) if parts[3] else 0 - chg = float(parts[32]) if parts[32] else 0 - return price, chg if price > 0 else (None, None) + c = str(code).strip().split("_")[0] + if c.lower().startswith("hk"): + c = c[2:] + conn = sqlite3.connect(DB_PATH, timeout=5) + row = conn.execute( + "SELECT price, change_pct FROM live_prices WHERE code=?", (c,) + ).fetchone() + conn.close() + if row and row[0]: + price = float(row[0]) + chg = float(row[1] or 0) + return (price, chg) if price > 0 else (None, None) except: pass return None, None diff --git a/deploy/profile-scripts/strategy_review.py b/deploy/profile-scripts/strategy_review.py index 50bd92f6..84bd8653 100644 --- a/deploy/profile-scripts/strategy_review.py +++ b/deploy/profile-scripts/strategy_review.py @@ -144,29 +144,32 @@ def evaluate_strategy(s, price): sl_recovery = False if tp > 0 or sl > 0: try: - prefix = "sh" if code.startswith(('60','68','51','56','50')) else "sz" if code.startswith(('00','30','15')) else "hk" - url = f"http://ifzq.gtimg.cn/appstock/app/fqkline/get?param={prefix}{code},day,,,60,qfq" - import subprocess as sp - r = sp.run(["curl", "-s", "--max-time", "3", url], capture_output=True, text=True, timeout=5) - if r.returncode == 0 and r.stdout: - data = json.loads(r.stdout) - day_key = 'qfqday' if prefix != 'hk' else 'day' - bars = data.get('data', {}).get(f'{prefix}{code}', {}).get(day_key, []) - if bars: - prices = [(float(b[2]), float(b[3]), b[0]) for b in bars if len(b) > 3] # (high, low, date) - recent_high = max(p[0] for p in prices) - recent_low = min(p[1] for p in prices) - # 检查止损触发后的走势:是否后来反弹了? - if sl > 0: - # 找出价格低于SL的K线 - below_sl = [p for p in prices if p[1] <= sl] - above_sl_later = [p for p in prices if p[1] > sl * 1.03] - if below_sl and above_sl_later: - # 曾跌破SL,但后来涨回去了 → 洗盘 - first_below = min(below_sl, key=lambda x: x[2]) - last_above = max(above_sl_later, key=lambda x: x[2]) - if last_above[2] > first_below[2]: - sl_recovery = True + # 读 stock_daily 近60根日K(2026-08-26 分层铁律:消费层不直连腾讯API) + raw_code = str(code).split("_")[0] + if raw_code.lower().startswith("hk"): + raw_code = raw_code[2:] + _conn = sqlite3.connect(str(DB_PATH), timeout=5) + _bars = _conn.execute( + "SELECT date, high, low FROM stock_daily WHERE code=? " + "ORDER BY date DESC LIMIT 60", (raw_code,) + ).fetchall() + _conn.close() + if _bars: + # 转为升序(旧→新),与 qfqday 时序一致 + prices = [(float(b[1]), float(b[2]), b[0]) for b in reversed(_bars) if b[1] and b[2]] # (high, low, date) + recent_high = max(p[0] for p in prices) + recent_low = min(p[1] for p in prices) + # 检查止损触发后的走势:是否后来反弹了? + if sl > 0: + # 找出价格低于SL的K线 + below_sl = [p for p in prices if p[1] <= sl] + above_sl_later = [p for p in prices if p[1] > sl * 1.03] + if below_sl and above_sl_later: + # 曾跌破SL,但后来涨回去了 → 洗盘 + first_below = min(below_sl, key=lambda x: x[2]) + last_above = max(above_sl_later, key=lambda x: x[2]) + if last_above[2] > first_below[2]: + sl_recovery = True except: pass