92 lines
3.2 KiB
Python
92 lines
3.2 KiB
Python
#!/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），EMA 平滑——与 backtest_framework.calc_atr 完全一致
|
||
（2026-08-12 统一口径：原 SMA 与回测 EMA 差 3.59%，回测实盘口径分裂）。
|
||
klines: [{high, low, close, ...}]。返回最新 ATR 值（单值）。"""
|
||
if len(klines) < n + 1:
|
||
return None
|
||
highs = [k["high"] for k in klines]
|
||
lows = [k["low"] for k in klines]
|
||
closes = [k["close"] for k in klines]
|
||
# True Range（与 backtest_framework.calc_tr 一致）
|
||
tr = [highs[0] - lows[0]]
|
||
for i in range(1, len(highs)):
|
||
hl = highs[i] - lows[i]
|
||
hc = abs(highs[i] - closes[i - 1])
|
||
lc = abs(lows[i] - closes[i - 1])
|
||
tr.append(max(hl, hc, lc))
|
||
# EMA 平滑（与 backtest_framework.calc_atr 一致：calc_ema(tr, n)）
|
||
k = 2 / (n + 1)
|
||
ema = tr[0]
|
||
for t in tr[1:]:
|
||
ema = t * k + ema * (1 - k)
|
||
return ema
|
||
|
||
|
||
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
|