87 lines
2.7 KiB
Python
87 lines
2.7 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
技术指标：EMA、ATR
|
||
"""
|
||
|
||
from typing import List, Dict, Optional
|
||
|
||
|
||
def ema(series: List[float], period: int) -> List[Optional[float]]:
|
||
"""EMA(period)，前 period-1 个为 None"""
|
||
if not series or period < 1:
|
||
return []
|
||
k = 2.0 / (period + 1)
|
||
out: List[Optional[float]] = [None] * (period - 1)
|
||
s = sum(series[:period])
|
||
out.append(s / period)
|
||
for i in range(period, len(series)):
|
||
val = series[i] * k + out[-1] * (1 - k) # type: ignore
|
||
out.append(val)
|
||
return out
|
||
|
||
|
||
def atr(high: List[float], low: List[float], close: List[float], period: int) -> List[Optional[float]]:
|
||
"""ATR(period)，前 period 个为 None"""
|
||
n = len(close)
|
||
if n < period + 1 or len(high) != n or len(low) != n:
|
||
return [None] * n
|
||
tr_list: List[float] = []
|
||
for i in range(n):
|
||
if i == 0:
|
||
tr_list.append(high[0] - low[0])
|
||
else:
|
||
tr = max(
|
||
high[i] - low[i],
|
||
abs(high[i] - close[i - 1]),
|
||
abs(low[i] - close[i - 1])
|
||
)
|
||
tr_list.append(tr)
|
||
return ema(tr_list, period)
|
||
|
||
|
||
def get_ema_atr_from_klines(klines: List[Dict], ema_period: int, atr_period: int
|
||
) -> tuple:
|
||
"""
|
||
从K线列表计算收盘价 EMA 和 ATR。
|
||
返回 (ema_list, atr_list)，长度与 klines 一致。
|
||
"""
|
||
close = [float(k["close"]) for k in klines]
|
||
high = [float(k["high"]) for k in klines]
|
||
low = [float(k["low"]) for k in klines]
|
||
ema_list = ema(close, ema_period)
|
||
atr_list = atr(high, low, close, atr_period)
|
||
return ema_list, atr_list
|
||
|
||
|
||
def align_higher_tf_ema(
|
||
klines_5m: List[Dict],
|
||
klines_higher: List[Dict],
|
||
ema_fast: int,
|
||
ema_slow: int
|
||
) -> List[Dict]:
|
||
"""
|
||
根据更高周期K线计算 EMA，并按 5 分钟时间对齐。
|
||
返回列表长度与 klines_5m 一致，每项为 {"ema_fast": float, "ema_slow": float} 或 None。
|
||
"""
|
||
if not klines_higher or len(klines_higher) < ema_slow:
|
||
return [{}] * len(klines_5m)
|
||
close_hi = [float(k["close"]) for k in klines_higher]
|
||
ema_f = ema(close_hi, ema_fast)
|
||
ema_s = ema(close_hi, ema_slow)
|
||
id_hi = [k["id"] for k in klines_higher]
|
||
result = []
|
||
for k5 in klines_5m:
|
||
t = k5["id"]
|
||
# 当前 5m 时刻所属的更高周期：取 <= t 的最后一根
|
||
idx = -1
|
||
for i, tid in enumerate(id_hi):
|
||
if tid <= t:
|
||
idx = i
|
||
else:
|
||
break
|
||
if idx >= ema_slow - 1 and ema_f[idx] is not None and ema_s[idx] is not None:
|
||
result.append({"ema_fast": ema_f[idx], "ema_slow": ema_s[idx]})
|
||
else:
|
||
result.append({})
|
||
return result
|