| from typing import Dict, List |
|
|
| import numpy as np |
| import pandas as pd |
| from ta.trend import ema_indicator, sma_indicator |
|
|
|
|
| def _finite_or_zero(value: float) -> float: |
| try: |
| value = float(value) |
| except Exception: |
| return 0.0 |
| if not np.isfinite(value): |
| return 0.0 |
| return value |
|
|
|
|
| def compute_rolling_quant_features(closes: List[float], end_idx: int) -> Dict[str, float]: |
| closes_np = np.asarray(closes[: end_idx + 1], dtype=np.float64) |
| if closes_np.size == 0: |
| return { |
| "ema_fast": 0.0, |
| "ema_medium": 0.0, |
| "sma_fast": 0.0, |
| "sma_medium": 0.0, |
| "price_minus_ema_fast": 0.0, |
| "price_minus_ema_medium": 0.0, |
| "ema_spread": 0.0, |
| "price_zscore": 0.0, |
| "mean_reversion_score": 0.0, |
| "rolling_vol_zscore": 0.0, |
| } |
|
|
| close_series = pd.Series(closes_np) |
| current = float(closes_np[-1]) |
| current_scale = max(abs(current), 1e-8) |
|
|
| ema_fast = _finite_or_zero(ema_indicator(close_series, window=8, fillna=True).iloc[-1]) |
| ema_medium = _finite_or_zero(ema_indicator(close_series, window=21, fillna=True).iloc[-1]) |
| sma_fast = _finite_or_zero(sma_indicator(close_series, window=8, fillna=True).iloc[-1]) |
| sma_medium = _finite_or_zero(sma_indicator(close_series, window=21, fillna=True).iloc[-1]) |
|
|
| mean_all = _finite_or_zero(close_series.mean()) |
| std_all = _finite_or_zero(close_series.std(ddof=0)) |
| price_zscore = 0.0 if std_all <= 1e-8 else (current - mean_all) / std_all |
|
|
| log_returns = np.diff(np.log(np.clip(closes_np, 1e-8, None))) |
| if log_returns.size == 0: |
| rolling_vol = 0.0 |
| mean_vol = 0.0 |
| std_vol = 0.0 |
| else: |
| abs_log_returns = np.abs(log_returns) |
| rolling_vol = _finite_or_zero(np.std(log_returns[-20:])) |
| mean_vol = _finite_or_zero(np.mean(abs_log_returns)) |
| std_vol = _finite_or_zero(np.std(abs_log_returns)) |
| rolling_vol_zscore = 0.0 if std_vol <= 1e-8 else (rolling_vol - mean_vol) / std_vol |
|
|
| denom = max(abs(sma_medium), 1e-8) |
| return { |
| "ema_fast": ema_fast / current_scale, |
| "ema_medium": ema_medium / current_scale, |
| "sma_fast": sma_fast / current_scale, |
| "sma_medium": sma_medium / current_scale, |
| "price_minus_ema_fast": (current - ema_fast) / current_scale, |
| "price_minus_ema_medium": (current - ema_medium) / current_scale, |
| "ema_spread": (ema_fast - ema_medium) / current_scale, |
| "price_zscore": _finite_or_zero(price_zscore), |
| "mean_reversion_score": (mean_all - current) / denom, |
| "rolling_vol_zscore": _finite_or_zero(rolling_vol_zscore), |
| } |
|
|
