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

refactor: 工具函数抽取公共模块——indicators.py(指标)+market_data.py(数据),mr/s2扫描器改为公共模块import(消除策略扫描器互import)

Browse Source
This commit is contained in:
hmo
2026-08-11 08:20:09 +08:00
parent af7419bc77
commit 0ef84d34d5
4 changed files with 171 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/profile-scripts/indicators.py
+81
View File
@@ -0,0 +1,81 @@
#!/usr/bin/env python3
"""indicators.py — 通用技术指标库（2026-08-11 从 mr_scanner 抽取）
背景：calc_ma/calc_rsi/calc_atr/calc_obv 原定义在 mr_scanner.py，
被 s2_scanner.py import 复用 —— 策略扫描器互相 import 工具函数是坏味道。
抽取到公共模块，供所有策略扫描器（mr/s2/accumulation/p_oversold）共用。
算法与 backtest_framework.py 完全一致（零偏差）。
"""
def calc_ma(series, n):
"""简单移动平均。前 n-1 位返回 None。"""
result = []
for i in range(len(series)):
if i < n - 1:
result.append(None)
else:
result.append(sum(series[i - n + 1:i + 1]) / n)
return result
def calc_rsi(series, n=14):
"""RSI（相对强弱指数），Wilder 平滑。"""
deltas = [series[i] - series[i - 1] for i in range(1, len(series))]
gains = [d if d > 0 else 0 for d in deltas]
losses = [-d if d < 0 else 0 for d in deltas]
result = [None] * (n + 1)
avg_gain = sum(gains[:n]) / n
avg_loss = sum(losses[:n]) / n
if avg_loss == 0:
result.append(100)
else:
rs = avg_gain / avg_loss
result.append(100 - 100 / (1 + rs))
for i in range(n, len(gains)):
avg_gain = (avg_gain * (n - 1) + gains[i]) / n
avg_loss = (avg_loss * (n - 1) + losses[i]) / n
if avg_loss == 0:
result.append(100)
else:
rs = avg_gain / avg_loss
result.append(100 - 100 / (1 + rs))
while len(result) < len(series):
result.insert(0, None)
return result[:len(series)]
def calc_atr(klines, n=14):
"""ATR（Average True Range），与回测 calc_factors 的 atr_pct 同口径。
klines: [{high, low, close, ...}]"""
if len(klines) < n + 1:
return None
trs = []
for i in range(1, len(klines)):
h, l, pc = klines[i]["high"], klines[i]["low"], klines[i - 1]["close"]
tr = max(h - l, abs(h - pc), abs(l - pc))
trs.append(tr)
atr = sum(trs[-n:]) / n
return atr
def calc_obv(klines):
"""OBV 能量潮（近20日变化量），资金流向指标。
klines: [{close, volume, ...}]"""
if len(klines) < 21:
return 0
obv = 0
for i in range(1, len(klines)):
if klines[i]["close"] > klines[i - 1]["close"]:
obv += klines[i]["volume"] * 100 # 手→股
elif klines[i]["close"] < klines[i - 1]["close"]:
obv -= klines[i]["volume"] * 100
# 近20日 OBV 变化
obv_now = 0
for i in range(max(1, len(klines) - 20), len(klines)):
if klines[i]["close"] > klines[i - 1]["close"]:
obv_now += klines[i]["volume"] * 100
elif klines[i]["close"] < klines[i - 1]["close"]:
obv_now -= klines[i]["volume"] * 100
return obv_now
deploy/profile-scripts/market_data.py
+80
View File
@@ -0,0 +1,80 @@
#!/usr/bin/env python3
"""market_data.py — 通用行情数据获取库（2026-08-11 从 mr_scanner 抽取）
背景：fetch_tx_klines/get_stock_pool 原定义在 mr_scanner.py，
被 s2_scanner.py import 复用 —— 策略扫描器互相 import 数据函数是坏味道。
抽取到公共模块，供所有策略扫描器（mr/s2/accumulation/p_oversold）共用。
"""
import json
import sqlite3
import urllib.request
from pathlib import Path
DB_PATH = Path("/home/hmo/MoFin/data/mofin.db")
UA = "Mozilla/5.0"
def fetch_tx_klines(code, datalen=120):
"""腾讯前复权日K（qfq），与 stock_daily 数据零偏差，返回 [{date,open,close,high,low,volume}]"""
raw = str(code).strip()
if raw.startswith(("6", "9")):
prefix = "sh"
elif raw.startswith(("0", "3")):
prefix = "sz"
else:
return None
url = f"http://ifzq.gtimg.cn/appstock/app/fqkline/get?param={prefix}{raw},day,,,{datalen},qfq"
try:
req = urllib.request.Request(url, headers={"User-Agent": UA})
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
with opener.open(req, timeout=8) as r:
text = r.read().decode("utf-8", errors="replace").strip()
data = json.loads(text)
node = data.get("data", {}).get(f"{prefix}{raw}", {})
bars = node.get("qfqday") or node.get("day") or []
if not bars or len(bars) < 70:
return None
result = []
for b in bars:
if len(b) < 6:
continue
result.append({
"date": b[0][:10],
"open": float(b[1]),
"close": float(b[2]),
"high": float(b[3]),
"low": float(b[4]),
"volume": float(b[5]), # 手
})
return result
except Exception:
return None
# 兼容别名（供外部引用）
fetch_sina_klines = fetch_tx_klines
def get_stock_pool():
"""待扫描股票池：stock_daily 的 distinct code（与回测 run_mr_backtest 完全同口径）
回测股票池 = SELECT DISTINCT sd.code FROM stock_daily（4266只，含300/688，
不含301新创业板——数据源未收录）。实盘扫描用同一口径，保证信号
覆盖的股票都是回测验证过的。
"""
conn = sqlite3.connect(str(DB_PATH), timeout=5)
try:
existing = set()
for r in conn.execute("SELECT code FROM holding_strategies WHERE status='active'"):
existing.add(str(r[0]))
for r in conn.execute("SELECT code FROM holdings WHERE is_active=1"):
existing.add(str(r[0]))
# 与回测完全一致：stock_daily 有K线的股票（回测 universe='a' 排除5位港股）
all_stocks = [str(r[0]) for r in
conn.execute("SELECT DISTINCT code FROM stock_daily").fetchall()]
finally:
conn.close()
# 只留 A 股（6位数字），排除港股（5位0开头）—— 与回测 is_hk_code 逻辑一致
a_stocks = [c for c in all_stocks if len(c) == 6 and c.isdigit()]
return a_stocks, existing
deploy/profile-scripts/mr_scanner.py
+7 -137
View File
@@ -36,6 +36,13 @@ import sys, json, urllib.request, re, time, sqlite3
from pathlib import Path
from datetime import datetime
# 2026-08-11 重构：工具函数抽取到公共模块（indicators/market_data）
# 原 calc_ma/calc_rsi/calc_atr/calc_obv → indicators.py
# 原 fetch_tx_klines/get_stock_pool → market_data.py
# 保留本文件的 import 兼容（从公共模块导入同名函数）
from indicators import calc_ma, calc_rsi, calc_atr, calc_obv
from market_data import fetch_tx_klines, get_stock_pool
DB_PATH = Path("/home/hmo/MoFin/data/mofin.db")
UA = "Mozilla/5.0"
@@ -55,143 +62,6 @@ EXIT_CFG = {"tp_pct": 0.30, "sl_pct": 0.12, "max_hold_days": 40}
TOP_N = 10
# ── 技术指标（与 backtest_framework.py 完全同算法）──
def calc_ma(series, n):
result = []
for i in range(len(series)):
if i < n - 1:
result.append(None)
else:
result.append(sum(series[i - n + 1:i + 1]) / n)
return result
def calc_rsi(series, n=14):
deltas = [series[i] - series[i - 1] for i in range(1, len(series))]
gains = [d if d > 0 else 0 for d in deltas]
losses = [-d if d < 0 else 0 for d in deltas]
result = [None] * (n + 1)
avg_gain = sum(gains[:n]) / n
avg_loss = sum(losses[:n]) / n
if avg_loss == 0:
result.append(100)
else:
rs = avg_gain / avg_loss
result.append(100 - 100 / (1 + rs))
for i in range(n, len(gains)):
avg_gain = (avg_gain * (n - 1) + gains[i]) / n
avg_loss = (avg_loss * (n - 1) + losses[i]) / n
if avg_loss == 0:
result.append(100)
else:
rs = avg_gain / avg_loss
result.append(100 - 100 / (1 + rs))
while len(result) < len(series):
result.insert(0, None)
return result[:len(series)]
def calc_atr(klines, n=14):
"""ATR（Average True Range），与回测 calc_factors 的 atr_pct 同口径"""
if len(klines) < n + 1:
return None
trs = []
for i in range(1, len(klines)):
h, l, pc = klines[i]["high"], klines[i]["low"], klines[i - 1]["close"]
tr = max(h - l, abs(h - pc), abs(l - pc))
trs.append(tr)
atr = sum(trs[-n:]) / n
return atr
def calc_obv(klines):
"""OBV 能量潮（近20日变化量），资金流向指标"""
if len(klines) < 21:
return 0
obv = 0
for i in range(1, len(klines)):
if klines[i]["close"] > klines[i - 1]["close"]:
obv += klines[i]["volume"] * 100 # 手→股
elif klines[i]["close"] < klines[i - 1]["close"]:
obv -= klines[i]["volume"] * 100
# 近20日 OBV 变化
obv_now = 0
for i in range(max(1, len(klines) - 20), len(klines)):
if klines[i]["close"] > klines[i - 1]["close"]:
obv_now += klines[i]["volume"] * 100
elif klines[i]["close"] < klines[i - 1]["close"]:
obv_now -= klines[i]["volume"] * 100
return obv_now
# ── 数据获取 ──
def fetch_tx_klines(code, datalen=120):
"""腾讯前复权日K（qfq），与 stock_daily 数据零偏差，返回 [{date,open,close,high,low,volume}]"""
raw = str(code).strip()
if raw.startswith(("6", "9")):
prefix = "sh"
elif raw.startswith(("0", "3")):
prefix = "sz"
else:
return None
url = f"http://ifzq.gtimg.cn/appstock/app/fqkline/get?param={prefix}{raw},day,,,{datalen},qfq"
try:
req = urllib.request.Request(url, headers={"User-Agent": UA})
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
with opener.open(req, timeout=8) as r:
text = r.read().decode("utf-8", errors="replace").strip()
data = json.loads(text)
node = data.get("data", {}).get(f"{prefix}{raw}", {})
bars = node.get("qfqday") or node.get("day") or []
if not bars or len(bars) < 70:
return None
result = []
for b in bars:
if len(b) < 6:
continue
result.append({
"date": b[0][:10],
"open": float(b[1]),
"close": float(b[2]),
"high": float(b[3]),
"low": float(b[4]),
"volume": float(b[5]), # 手
})
return result
except Exception:
return None
# 兼容别名（供外部引用）
fetch_sina_klines = fetch_tx_klines
def get_stock_pool():
"""待扫描股票池：stock_daily 的 distinct code（与回测 run_mr_backtest 完全同口径）
回测股票池 = SELECT DISTINCT sd.code FROM stock_daily（4266只，含300/688，
不含301新创业板——数据源未收录）。实盘扫描用同一口径，保证 v_mr
信号覆盖的股票都是回测验证过的。
"""
conn = sqlite3.connect(str(DB_PATH), timeout=5)
try:
existing = set()
for r in conn.execute("SELECT code FROM holding_strategies WHERE status='active'"):
existing.add(str(r[0]))
for r in conn.execute("SELECT code FROM holdings WHERE is_active=1"):
existing.add(str(r[0]))
# 与回测完全一致：stock_daily 有K线的股票（回测 universe='a' 排除5位港股）
all_stocks = [str(r[0]) for r in
conn.execute("SELECT DISTINCT code FROM stock_daily").fetchall()]
finally:
conn.close()
# 只留 A 股（6位数字），排除港股（5位0开头）—— 与回测 is_hk_code 逻辑一致
a_stocks = [c for c in all_stocks if len(c) == 6 and c.isdigit()]
return a_stocks, existing
def load_regime():
"""读取 market_regime 最新状态"""
try:
deploy/profile-scripts/s2_scanner.py
+3 -1
View File
@@ -30,7 +30,9 @@ from pathlib import Path
from datetime import datetime
sys.path.insert(0, str(Path(__file__).parent))
from mr_scanner import fetch_tx_klines, calc_ma, calc_rsi, get_stock_pool
# 2026-08-11 重构：工具函数从 mr_scanner 抽到公共模块，s2 直接引用公共模块
from indicators import calc_ma, calc_rsi
from market_data import fetch_tx_klines, get_stock_pool
DB_PATH = Path("/home/hmo/MoFin/data/mofin.db")