Spaces:
Running
Running
| """ | |
| utils/normalizer.py | |
| Central payload normalizer for live scanner and PA scanner outputs. | |
| Purpose: | |
| - prevent UI crashes from mixed payload shapes | |
| - normalize dict/string/list fields | |
| - fill missing nested keys with safe defaults | |
| """ | |
| from __future__ import annotations | |
| from copy import deepcopy | |
| from typing import Any, Dict, List | |
| def _safe_dict(v: Any) -> Dict[str, Any]: | |
| return v if isinstance(v, dict) else {} | |
| def _safe_list(v: Any) -> List[Any]: | |
| return v if isinstance(v, list) else [] | |
| def _safe_float(v: Any, default: float | None = None) -> float | None: | |
| try: | |
| if v is None or v == "": | |
| return default | |
| return float(v) | |
| except Exception: | |
| return default | |
| def _safe_int(v: Any, default: int = 0) -> int: | |
| try: | |
| if v is None or v == "": | |
| return default | |
| return int(v) | |
| except Exception: | |
| return default | |
| def _normalize_regime(v: Any) -> Dict[str, Any]: | |
| if isinstance(v, dict): | |
| return { | |
| "regime": v.get("regime", v.get("label", "unknown")), | |
| "label": v.get("label", v.get("regime", "unknown")), | |
| } | |
| if isinstance(v, str): | |
| return {"regime": v, "label": v} | |
| return {"regime": "unknown", "label": "unknown"} | |
| def _normalize_setup(v: Any) -> Dict[str, Any]: | |
| if isinstance(v, dict): | |
| return {"name": v.get("name", "No clear setup"), **v} | |
| if isinstance(v, str): | |
| return {"name": v} | |
| return {"name": "No clear setup"} | |
| def _normalize_levels(v: Any) -> Dict[str, Any]: | |
| d = _safe_dict(v) | |
| return { | |
| "entry": d.get("entry"), | |
| "sl": d.get("sl"), | |
| "tp": d.get("tp"), | |
| "tp1": d.get("tp1"), | |
| "tp2": d.get("tp2"), | |
| "tp3": d.get("tp3"), | |
| "be": d.get("be"), | |
| "support": d.get("support"), | |
| "resistance": d.get("resistance"), | |
| "rr_ratio": d.get("rr_ratio"), | |
| "good_rr": bool(d.get("good_rr", False)), | |
| "room_pips": d.get("room_pips"), | |
| "sl_pips": d.get("sl_pips"), | |
| "tp_pips": d.get("tp_pips"), | |
| "tp1_pips": d.get("tp1_pips"), | |
| "tp2_pips": d.get("tp2_pips"), | |
| "tp3_pips": d.get("tp3_pips"), | |
| "risk_pips": d.get("risk_pips"), | |
| "reward_pips": d.get("reward_pips"), | |
| "sl_source": d.get("sl_source", "n/a"), | |
| "tp_source": d.get("tp_source", "n/a"), | |
| "digits": _safe_int(d.get("digits"), 5), | |
| "atr_pips": d.get("atr_pips"), | |
| "spread": d.get("spread"), | |
| **d, | |
| } | |
| def _normalize_patterns(v: Any) -> Dict[str, Any]: | |
| d = _safe_dict(v) | |
| return { | |
| "patterns": _safe_dict(d.get("patterns")), | |
| "pattern_score": _safe_float(d.get("pattern_score"), 0.0), | |
| "any_pattern": bool(d.get("any_pattern", False)), | |
| "best": d.get("best"), | |
| **d, | |
| } | |
| def _normalize_session(v: Any) -> Dict[str, Any]: | |
| d = _safe_dict(v) | |
| return { | |
| "allowed": bool(d.get("allowed", False)), | |
| "sessions": _safe_list(d.get("sessions")), | |
| "active": _safe_list(d.get("active")), | |
| "overlaps": _safe_list(d.get("overlaps")), | |
| "score": d.get("score", 0), | |
| "reason": d.get("reason", d.get("note", "")), | |
| "note": d.get("note", d.get("reason", "")), | |
| **d, | |
| } | |
| def _normalize_news(v: Any) -> Dict[str, Any]: | |
| d = _safe_dict(v) | |
| blocking = d.get("blocking", []) | |
| if not isinstance(blocking, list): | |
| blocking = [blocking] | |
| return { | |
| "allowed": bool(d.get("allowed", True)), | |
| "ok": bool(d.get("ok", d.get("allowed", True))), | |
| "note": d.get("note", "Clear"), | |
| "events": _safe_list(d.get("events")), | |
| "blocking": blocking, | |
| "score": d.get("score", 1.0), | |
| **d, | |
| } | |
| def _normalize_tf_block(v: Any, include_bias: bool = True) -> Dict[str, Any]: | |
| d = _safe_dict(v) | |
| out = { | |
| "rsi": d.get("rsi"), | |
| "adx": d.get("adx"), | |
| "zone": d.get("zone", "mixed"), | |
| } | |
| if include_bias: | |
| out.update( | |
| { | |
| "bias": d.get("bias", 0), | |
| "label": d.get("label", "neutral"), | |
| "stack_bull": bool(d.get("stack_bull", False)), | |
| "stack_bear": bool(d.get("stack_bear", False)), | |
| "ema20": d.get("ema20"), | |
| "ema50": d.get("ema50"), | |
| "sma200": d.get("sma200"), | |
| "sma_slope": d.get("sma_slope", "?"), | |
| } | |
| ) | |
| out.update(d) | |
| return out | |
| def _normalize_ltf(v: Any) -> Dict[str, Any]: | |
| d = _safe_dict(v) | |
| return { | |
| "ltf": d.get("ltf", "H1"), | |
| "rsi": d.get("rsi"), | |
| "atr": d.get("atr"), | |
| "adx": d.get("adx"), | |
| "triggered": bool(d.get("triggered", False)), | |
| "signal": d.get("signal", 0), | |
| "cross_ok": bool(d.get("cross_ok", False)), | |
| "rsi_ok": bool(d.get("rsi_ok", False)), | |
| "zone": d.get("zone", "mixed"), | |
| "rsi_mode": d.get("rsi_mode", "pullback"), | |
| "trigger_mode": d.get("trigger_mode", "none"), | |
| "trigger_note": d.get("trigger_note", ""), | |
| "entry_value_ok": bool(d.get("entry_value_ok", True)), | |
| "entry_value_note": d.get("entry_value_note", ""), | |
| "rsi_phase": d.get("rsi_phase", "unknown"), | |
| "value_score": d.get("value_score"), | |
| "value_score_note": d.get("value_score_note", ""), | |
| "entry_timing_score": d.get("entry_timing_score"), | |
| "entry_timing_note": d.get("entry_timing_note", ""), | |
| **d, | |
| } | |
| def _normalize_explanation(v: Any) -> Dict[str, Any]: | |
| d = _safe_dict(v) | |
| return { | |
| "summary": d.get("summary", ""), | |
| "state": d.get("state", ""), | |
| "bias_story": d.get("bias_story", ""), | |
| "location_story": d.get("location_story", ""), | |
| "mtf_story": d.get("mtf_story", ""), | |
| "mtf_checks_verbose": _safe_dict(d.get("mtf_checks_verbose")), | |
| "ltf_story": d.get("ltf_story", ""), | |
| "trigger_expectation": d.get("trigger_expectation", ""), | |
| "confirmation_story": d.get("confirmation_story", ""), | |
| "momentum_story": d.get("momentum_story", ""), | |
| "risk_story": d.get("risk_story", ""), | |
| "session_story": d.get("session_story", ""), | |
| "news_story": d.get("news_story", ""), | |
| "blockers_story": d.get("blockers_story", ""), | |
| "waiting_story": d.get("waiting_story", ""), | |
| "next_step": d.get("next_step", ""), | |
| **d, | |
| } | |
| def normalize_signal_payload(result: Dict[str, Any]) -> Dict[str, Any]: | |
| """ | |
| Normalize one scanner result payload into a stable shape for UI/audit. | |
| """ | |
| r = deepcopy(result) if isinstance(result, dict) else {} | |
| r["setup"] = _normalize_setup(r.get("setup")) | |
| r["regime"] = _normalize_regime(r.get("regime")) | |
| if "regime_v2" in r: | |
| r["regime_v2"] = _normalize_regime(r.get("regime_v2")) | |
| r["levels"] = _normalize_levels(r.get("levels")) | |
| r["patterns"] = _normalize_patterns(r.get("patterns")) | |
| r["session"] = _normalize_session(r.get("session")) | |
| r["news"] = _normalize_news(r.get("news")) | |
| r["wtf"] = _normalize_tf_block(r.get("wtf"), include_bias=True) | |
| r["htf"] = _normalize_tf_block(r.get("htf"), include_bias=True) | |
| mtf = _safe_dict(r.get("mtf")) | |
| r["mtf"] = { | |
| "zone": mtf.get("zone", "mixed"), | |
| "zone_label": mtf.get("zone_label", "mixed"), | |
| "sr_levels": _safe_list(mtf.get("sr_levels")), | |
| "score": _safe_int(mtf.get("score"), 0), | |
| "max": _safe_int(mtf.get("max"), 0), | |
| "confirmed": bool(mtf.get("confirmed", False)), | |
| "rsi": mtf.get("rsi"), | |
| "adx": mtf.get("adx"), | |
| "checks": _safe_dict(mtf.get("checks")), | |
| **mtf, | |
| } | |
| r["ltf"] = _normalize_ltf(r.get("ltf")) | |
| r["explanation"] = _normalize_explanation(r.get("explanation")) | |
| r["blockers"] = [str(x) for x in _safe_list(r.get("blockers"))] | |
| r["waiting_for"] = [str(x) for x in _safe_list(r.get("waiting_for"))] | |
| r["signal"] = _safe_int(r.get("signal"), 0) | |
| r["tradeable"] = bool(r.get("tradeable", False)) | |
| r["grade"] = str(r.get("grade", "NO TRADE")) | |
| r["direction"] = str(r.get("direction", "—")) | |
| r["decision"] = str(r.get("decision", "NO TRADE")) | |
| r["instrument"] = str(r.get("instrument", "—")) | |
| r["ltf_tf"] = str(r.get("ltf_tf", "H1")) | |
| # PA-specific optional blocks | |
| r["context"] = _safe_dict(r.get("context")) | |
| r["confirmed_setups"] = [str(x) for x in _safe_list(r.get("confirmed_setups"))] | |
| r["near_levels"] = _safe_dict(r.get("near_levels")) | |
| r["key_levels"] = _safe_dict(r.get("key_levels")) | |
| r["pa_setups"] = _safe_dict(r.get("pa_setups")) | |
| return r | |
| def normalize_signal_list(results: List[Dict[str, Any]]) -> List[Dict[str, Any]]: | |
| return [normalize_signal_payload(x) for x in results if isinstance(x, dict)] | |