130 lines
7.3 KiB
Python
130 lines
7.3 KiB
Python
"""Confirmed pivots, setup candidates and transparent deterministic focus scores."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from .config import DEFAULT_FOCUS_WEIGHTS
|
|
|
|
DEFAULT_SCORE_WEIGHTS = DEFAULT_FOCUS_WEIGHTS
|
|
|
|
|
|
def confirmed_pivots(low: pd.Series, *, left_bars: int = 3, right_bars: int = 3) -> pd.DataFrame:
|
|
"""A low pivot is published only on its confirmation bar, never its pivot bar."""
|
|
if left_bars < 1 or right_bars < 1:
|
|
raise ValueError("pivot windows must be positive")
|
|
values = low.to_numpy(dtype=float)
|
|
pivot_at_confirmation = np.full(len(values), np.nan)
|
|
last_pivot = np.full(len(values), np.nan)
|
|
higher_low = np.zeros(len(values), dtype=bool)
|
|
previous_pivot = np.nan
|
|
current_pivot = np.nan
|
|
is_higher = False
|
|
for i in range(len(values)):
|
|
candidate = i - right_bars
|
|
if candidate >= left_bars and i < len(values):
|
|
window = values[candidate - left_bars : candidate + right_bars + 1]
|
|
if np.isfinite(window).all() and values[candidate] == np.min(window):
|
|
# Equal lows are ambiguous; confirm only a unique minimum.
|
|
if int(np.sum(window == values[candidate])) == 1:
|
|
previous_pivot = current_pivot
|
|
current_pivot = values[candidate]
|
|
is_higher = np.isfinite(previous_pivot) and current_pivot > previous_pivot
|
|
pivot_at_confirmation[i] = current_pivot
|
|
if np.isfinite(current_pivot):
|
|
last_pivot[i] = current_pivot
|
|
higher_low[i] = is_higher
|
|
return pd.DataFrame({"confirmed_pivot_low": pivot_at_confirmation,
|
|
"last_confirmed_pivot_low": last_pivot,
|
|
"higher_low": higher_low}, index=low.index)
|
|
|
|
|
|
def candidate_features(features: pd.DataFrame, *, pullback_window: int = 5,
|
|
tightness_window: int = 10, volume_ratio: float = 1.0,
|
|
pivot_left: int = 3, pivot_right: int = 3) -> pd.DataFrame:
|
|
out = features.copy().sort_index()
|
|
pivots = confirmed_pivots(out["low"], left_bars=pivot_left, right_bars=pivot_right)
|
|
out[pivots.columns] = pivots
|
|
under_structure = out["close"] < out["ema21_high"]
|
|
pullback_volume = out["volume"].where((out["close"] < out["close"].shift(1)) | under_structure)
|
|
out["pullback_volume"] = pullback_volume.rolling(pullback_window, min_periods=1).mean()
|
|
out["volume_contracting"] = out["pullback_volume"] <= out["avg_volume20"] * volume_ratio
|
|
out["atr_contraction"] = out["atr14"] / out["atr14"].rolling(tightness_window, min_periods=max(3, tightness_window // 2)).mean().replace(0, np.nan)
|
|
rolling_range = out["high"].rolling(tightness_window, min_periods=max(3, tightness_window // 2)).max() - out["low"].rolling(tightness_window, min_periods=max(3, tightness_window // 2)).min()
|
|
out["range_contraction_pct"] = rolling_range / out["close"].replace(0, np.nan) * 100
|
|
out["close_volatility"] = out["close"].pct_change(fill_method=None).rolling(tightness_window, min_periods=max(3, tightness_window // 2)).std(ddof=0)
|
|
def trailing_percentile(values: np.ndarray) -> float:
|
|
return float(np.mean(values <= values[-1]))
|
|
range_rank = out["range_contraction_pct"].rolling(tightness_window, min_periods=max(3, tightness_window // 2)).apply(trailing_percentile, raw=True)
|
|
volatility_rank = out["close_volatility"].rolling(tightness_window, min_periods=max(3, tightness_window // 2)).apply(trailing_percentile, raw=True)
|
|
out["tightness_score"] = (
|
|
(1 - out["atr_contraction"].clip(0, 2) / 2) * 40
|
|
+ (1 - range_rank.fillna(.5)) * 30
|
|
+ (1 - volatility_rank.fillna(.5)) * 30
|
|
).clip(0, 100)
|
|
return out
|
|
|
|
|
|
def market_alignment_score(regime: str) -> float:
|
|
return {"CONFIRMED_UPTREND": 100, "UPTREND_PULLBACK": 100,
|
|
"EARLY_UPTREND": 75, "REPAIR": 55, "CORRECTION": 0,
|
|
"OVERBOUGHT": 20, "BREAKDOWN": 0}.get(str(regime).upper(), 0)
|
|
|
|
|
|
def focus_score(row: pd.Series, *, sector_strength: float | None = None,
|
|
weights: dict[str, float] | None = None) -> float:
|
|
weights = weights or DEFAULT_SCORE_WEIGHTS
|
|
extension = row.get("extension_atr", 0.0)
|
|
distance = abs(float(extension)) if pd.notna(extension) else 1.5
|
|
rs_value = row.get("rs_percentile", 0)
|
|
tightness_value = row.get("tightness_score", 0)
|
|
values = {
|
|
"rs": float(rs_value) if pd.notna(rs_value) else 0.0,
|
|
"market_alignment": market_alignment_score(row.get("regime", "CORRECTION")),
|
|
"structure": 100.0 if row.get("trend_state") == "UP" else 50.0 if row.get("trend_state") == "NEUTRAL" else 0.0,
|
|
"higher_low": 100.0 if bool(row.get("higher_low", False)) else 0.0,
|
|
"atr_distance": float(np.clip(100 * (1 - min(distance, 1.5) / 1.5), 0, 100)),
|
|
"volume_behavior": 100.0 if bool(row.get("volume_contracting", False)) else 0.0,
|
|
"tightness": float(tightness_value) if pd.notna(tightness_value) else 0.0,
|
|
"sector_strength": sector_strength,
|
|
"weekly": 100.0 if row.get("weekly_trend_state") == "UP" else 50.0 if row.get("weekly_trend_state") == "NEUTRAL" else 0.0,
|
|
}
|
|
active = [(values.get(key), weight) for key, weight in weights.items()
|
|
if values.get(key) is not None and pd.notna(values.get(key)) and weight > 0]
|
|
if not active:
|
|
return 0.0
|
|
total = sum(weight for _, weight in active)
|
|
return round(sum(float(value) * weight for value, weight in active) / total, 2)
|
|
|
|
|
|
def setup_for(row: pd.Series, previous: pd.Series | None = None, *,
|
|
prior_high: float | None = None,
|
|
max_extension_atr: float = 1.0) -> tuple[str | None, float | None, float | None]:
|
|
"""Return (setup, proposed entry reference, structural stop) on known daily data."""
|
|
close, low, high = (row.get(key, np.nan) for key in ("close", "low", "high"))
|
|
stop = row.get("ema21_low", np.nan)
|
|
if not all(pd.notna(v) for v in (close, low, high, stop)) or stop >= close:
|
|
return None, None, None
|
|
extension = row.get("extension_atr", 0.0)
|
|
if pd.notna(extension) and extension > max_extension_atr:
|
|
return None, None, None
|
|
touched = low <= row.get("ema21_high", np.inf) and close >= stop
|
|
if touched and row.get("trend_state") == "UP":
|
|
return "21DMA Retest", float(close), float(stop)
|
|
if bool(row.get("higher_low", False)) and close >= row.get("ema21_close", np.inf):
|
|
return "Higher Low", float(close), float(stop)
|
|
if previous is not None:
|
|
if prior_high is not None and pd.notna(prior_high) and close > prior_high:
|
|
return "Base Breakout", float(close), float(stop)
|
|
if close > previous.get("high", np.inf) and row.get("close", 0) > row.get("open", 0):
|
|
return "Prior High Reclaim", float(close), float(stop)
|
|
if (close > row.get("ema21_high", np.inf)
|
|
and previous.get("close", np.inf) <= previous.get("ema21_high", -np.inf)):
|
|
return "21DMA High Reclaim", float(close), float(stop)
|
|
if row.get("open", np.nan) < previous.get("close", np.inf) and close > previous.get("close", np.inf):
|
|
return "Red-to-Green", float(close), float(stop)
|
|
if close > row.get("open", np.inf) and previous.get("close", -np.inf) < previous.get("open", np.inf):
|
|
return "Daily Reversal", float(close), float(stop)
|
|
return None, None, None
|