"""
entry_rules.py -- trade1 ENTRY module built on the entry_rules engine
======================================================================

[2026-09-27] Installed in trade1 as a real entry module (rules_api
contract -- see the adapter at the bottom of this file): it BUYS when
the engine's verdict is PASS. The engine below is the same code that runs
as the shadow recorder in screener/trade (copied unchanged); its original
notes follow.

----------------------------------------------------------------------

[2026-09-26] Adapted from the sip_bot version the user supplied, for this
bot (stream.py / monitor.py / simulate.py). Changes, per the review:
  * record() builds the snapshot under the lock but writes the file AFTER
    releasing it, so the stream thread's on_trade/on_quote are never blocked
    by disk I/O (they share the lock).
  * track_outcomes (default True here, False in the live config): live, the
    outcomes are labeled after the close from Alpaca 1-min bars by
    shadow_outcomes.py (longer horizons, target-before-stop, and no memory
    held during the day or lost when a rescan unsubscribes a symbol).
  * seed_bars(): load a symbol's bars since the open in one go -- stocks added
    by a mid-day rescan otherwise start VWAP/HOD/EMA from the time they were
    added.
  * The caller (monitor.py) records only on decision CHANGES, not every
    5-second REJECTION.

Purpose
-------
Every time the bot evaluates an entry (ENTRY, REJECTION, CONFIRM_START, ...)
this module records a full feature snapshot, scores it with the *proposed*
entry rules (Volume Acceleration Score, order-flow pressure, relative
strength, spread, 3-bar volume trend), and later attaches what price actually
did afterwards (+1 / +5 / +15 min returns, MFE, MAE).

SHADOW ONLY:
  * never places orders and never changes a bot decision
  * never raises into the caller -- every public method is wrapped, internal
    errors are logged to logger "entry_rules" and swallowed
  * delete the hooks and the bot behaves exactly as before

Output
------
<log_dir>/shadow_entry_YYYY-MM-DD.jsonl  (one JSON object per line, ET date)
  {"type": "snapshot", "id": ..., ...}   written at decision time
  {"type": "outcome",  "id": ..., ...}   written once all horizons elapse
Join the two on "id" for analysis.

Integration (4 hooks)
---------------------
    from entry_rules import ShadowEntryRules, Bar

    shadow = ShadowEntryRules(config.get("shadow_entry_rules", {}))

    # stream handlers -- feed universe symbols AND benchmarks (SPY, IWM)
    shadow.on_bar(sym, Bar(ts, o, h, l, c, v))   # COMPLETED bar, ts = bar START
    shadow.on_quote(sym, ts, bid, ask, bid_size, ask_size)
    shadow.on_trade(sym, ts, price, size)

    # decision points (entry_engine)
    shadow.record(sym, "ENTRY", bot_context={"health": h, "momentum": m,
                                             "reason": r, "setup": s})
    shadow.record(sym, "REJECTION", bot_context={"reason": "rvol_floor"})

    # end of day (writes partial outcomes for anything still pending)
    shadow.flush()

All timestamps must be timezone-aware datetimes. Pass `now=` to record()
when replaying historical data; live, it defaults to the wall clock.

Notes
-----
* VWAP / HOD / EMAs / ATR are computed from the bars you feed. Feed RTH bars
  only if you want RTH values (premarket bars would move HOD and VWAP).
* Trade side uses the Lee-Ready rule (quote rule, then tick rule) against the
  most recent quote, so quotes and trades must be fed in arrival order.
* Outcome horizons are measured at the close of the first completed bar that
  ends at or after t0 + horizon. Bars that started before the snapshot are
  excluded, so MFE/MAE never include pre-entry prices.
"""
from __future__ import annotations

import functools
import json
import logging
import math
import os
import threading
import uuid
from collections import deque
from dataclasses import dataclass
from datetime import date, datetime, timedelta
from typing import Any, Deque, Dict, List, Optional, Tuple
from zoneinfo import ZoneInfo

log = logging.getLogger("entry_rules")
ET = ZoneInfo("America/New_York")

DEFAULT_CONFIG: Dict[str, Any] = {
    "enabled": True,
    "log_dir": "logs/shadow",
    "bar_seconds": 60,
    "max_bars": 600,
    # --- volume -----------------------------------------------------------
    "rvol_lookback_bars": 20,      # baseline for bar-level relative volume
    "min_baseline_bars": 5,        # fewer -> volume features are None
    "accel_window_bars": 3,        # smoothed accel: last 3 bars vs prior 3
    "raw_accel_floor_frac": 0.25,  # raw accel denominator floor (x baseline avg)
    "momentum_threshold": 0.10,    # accel change counted as rising / falling
    "min_projected_elapsed_s": 15, # in-progress bar projection needs >= 15 s
    # --- order flow -------------------------------------------------------
    "imbalance_window_s": 30,
    "max_quote_age_s": 2.0,        # older quotes ignored for trade side
    # --- context ----------------------------------------------------------
    "atr_period": 14,
    "rs_lookback_bars": 5,
    "benchmark_symbols": ["SPY", "IWM"],
    # --- outcomes ---------------------------------------------------------
    "outcome_horizons_s": [60, 300, 900],
    "track_outcomes": True,        # False: label offline after the close (shadow_outcomes.py)
    # --- scoring tables: [upper_bound, score], null upper = +infinity ------
    "rvol_buckets": [[0.75, -2], [1.0, -1], [1.5, 0], [2.0, 1], [3.0, 2], [None, 3]],
    "accel_buckets": [[-0.30, -2], [-0.10, -1], [0.10, 0], [0.30, 1], [0.60, 2], [None, 3]],
    "imbalance_buckets": [[-0.30, -2], [-0.10, -1], [0.10, 0], [0.30, 1], [None, 2]],
    # --- proposed rules (evaluated, never enforced) ------------------------
    "rules": {
        "min_vas": 6.0,
        "min_imbalance": 0.10,
        "min_bar_rvol": 1.0,
        "max_spread_pct": 0.50,
        "min_rel_strength_pct": 0.0,   # must beat EVERY benchmark
        "block_decelerating": True,
    },
}


@dataclass
class Bar:
    ts: datetime  # bar START, tz-aware
    open: float
    high: float
    low: float
    close: float
    volume: float


# ---------------------------------------------------------------------------
# helpers
# ---------------------------------------------------------------------------
def _merge(base: dict, override: dict) -> dict:
    out = dict(base)
    for k, v in override.items():
        if isinstance(v, dict) and isinstance(out.get(k), dict):
            out[k] = _merge(out[k], v)
        else:
            out[k] = v
    return out


def _safe(fn):
    @functools.wraps(fn)
    def wrapper(self, *args, **kwargs):
        if not self.cfg.get("enabled", True):
            return None
        try:
            return fn(self, *args, **kwargs)
        except Exception:  # shadow mode must never break the bot
            log.exception("entry_rules: error in %s (ignored)", fn.__name__)
            return None
    return wrapper


def _et_date(ts: datetime) -> date:
    return ts.astimezone(ET).date()


def _floor(ts: datetime, seconds: int) -> datetime:
    epoch = ts.timestamp()
    return datetime.fromtimestamp(epoch - (epoch % seconds), tz=ts.tzinfo)


def _ema(prev: Optional[float], value: float, n: int) -> float:
    if prev is None:
        return value
    a = 2.0 / (n + 1)
    return prev + a * (value - prev)


def _ratio_minus_one(a: float, b: float) -> Optional[float]:
    return a / b - 1.0 if b > 0 else None


def _pct(a: Optional[float], b: Optional[float]) -> Optional[float]:
    if a is None or b is None or b == 0:
        return None
    return (a / b - 1.0) * 100.0


def _bucket(value: Optional[float], table: List[list]) -> Optional[int]:
    if value is None:
        return None
    for upper, score in table:
        if upper is None or value < upper:
            return score
    return table[-1][1]


def _change_pct(bars: Deque[Bar], k: int) -> Optional[float]:
    if bars is None or len(bars) <= k:
        return None
    return _pct(bars[-1].close, bars[-1 - k].close)


def _clean(obj: Any) -> Any:
    """JSON-safe copy: datetimes -> ISO, NaN/inf -> None, floats rounded."""
    if isinstance(obj, dict):
        return {str(k): _clean(v) for k, v in obj.items()}
    if isinstance(obj, (list, tuple)):
        return [_clean(v) for v in obj]
    if isinstance(obj, datetime):
        return obj.isoformat()
    if isinstance(obj, bool) or obj is None or isinstance(obj, (int, str)):
        return obj
    if isinstance(obj, float):
        return None if math.isnan(obj) or math.isinf(obj) else round(obj, 6)
    return str(obj)


class _SymbolState:
    def __init__(self, day: date, max_bars: int):
        self.date = day
        self.bars: Deque[Bar] = deque(maxlen=max_bars)
        self.trades: Deque[Tuple[datetime, float, float, int]] = deque()
        self.quote: Optional[Tuple[datetime, float, float, float, float]] = None
        self.last_price: Optional[float] = None
        self.last_side = 0
        self.cur_bar_start: Optional[datetime] = None
        self.cur_bar_volume = 0.0
        self.cum_pv = 0.0
        self.cum_v = 0.0
        self.hod: Optional[float] = None
        self.ema9: Optional[float] = None
        self.ema20: Optional[float] = None
        self.atr: Optional[float] = None
        self.prev_close: Optional[float] = None


# ---------------------------------------------------------------------------
# main class
# ---------------------------------------------------------------------------
class ShadowEntryRules:
    def __init__(self, config: Optional[dict] = None):
        self.cfg = _merge(DEFAULT_CONFIG, config or {})
        self._lock = threading.RLock()
        self._states: Dict[str, _SymbolState] = {}
        self._pending: Dict[str, List[dict]] = {}
        self._bar_td = timedelta(seconds=self.cfg["bar_seconds"])

    # ------------------------------------------------------------ state
    def _state(self, symbol: str, ts: datetime) -> _SymbolState:
        """Per-symbol state, reset automatically on a new ET date."""
        day = _et_date(ts)
        st = self._states.get(symbol)
        if st is None or st.date != day:
            st = _SymbolState(day, self.cfg["max_bars"])
            self._states[symbol] = st
        return st

    # ------------------------------------------------------------ feeds
    @_safe
    def seed_bars(self, symbol: str, bars: List[Bar]) -> None:
        """Replace today's state for symbol with its bars since the open
        (oldest first), e.g. when a rescan adds it mid-session."""
        if not bars:
            return
        with self._lock:
            self._states.pop(symbol, None)
        for b in bars:
            self.on_bar(symbol, b)

    @_safe
    def on_bar(self, symbol: str, bar: Bar) -> None:
        with self._lock:
            st = self._state(symbol, bar.ts)
            st.bars.append(bar)
            typical = (bar.high + bar.low + bar.close) / 3.0
            st.cum_pv += typical * bar.volume
            st.cum_v += bar.volume
            st.hod = bar.high if st.hod is None else max(st.hod, bar.high)
            st.ema9 = _ema(st.ema9, bar.close, 9)
            st.ema20 = _ema(st.ema20, bar.close, 20)
            if st.prev_close is None:
                tr = bar.high - bar.low
            else:
                tr = max(bar.high - bar.low,
                         abs(bar.high - st.prev_close),
                         abs(bar.low - st.prev_close))
            n = self.cfg["atr_period"]
            st.atr = tr if st.atr is None else st.atr + (tr - st.atr) / n
            st.prev_close = bar.close
            self._update_outcomes(symbol, bar)

    @_safe
    def on_quote(self, symbol: str, ts: datetime, bid: float, ask: float,
                 bid_size: float = 0.0, ask_size: float = 0.0) -> None:
        with self._lock:
            st = self._state(symbol, ts)
            st.quote = (ts, float(bid), float(ask), float(bid_size), float(ask_size))

    @_safe
    def on_trade(self, symbol: str, ts: datetime, price: float, size: float) -> None:
        with self._lock:
            st = self._state(symbol, ts)
            price, size = float(price), float(size)
            side = self._classify(st, ts, price)
            st.last_price, st.last_side = price, side
            st.trades.append((ts, price, size, side))
            keep_s = max(self.cfg["imbalance_window_s"], self.cfg["bar_seconds"])
            cutoff = ts - timedelta(seconds=keep_s)
            while st.trades and st.trades[0][0] < cutoff:
                st.trades.popleft()
            start = _floor(ts, self.cfg["bar_seconds"])
            if st.cur_bar_start != start:
                st.cur_bar_start, st.cur_bar_volume = start, 0.0
            st.cur_bar_volume += size

    def _classify(self, st: _SymbolState, ts: datetime, price: float) -> int:
        """Lee-Ready: quote rule (ask/bid/midpoint), then tick rule."""
        q = st.quote
        if q is not None and (ts - q[0]).total_seconds() <= self.cfg["max_quote_age_s"]:
            _, bid, ask, _, _ = q
            if bid > 0 and ask >= bid:
                if price >= ask:
                    return 1
                if price <= bid:
                    return -1
                mid = (bid + ask) / 2.0
                if price > mid:
                    return 1
                if price < mid:
                    return -1
        if st.last_price is not None:
            if price > st.last_price:
                return 1
            if price < st.last_price:
                return -1
        return st.last_side

    # ------------------------------------------------------------ record
    @_safe
    def record(self, symbol: str, event: str, now: Optional[datetime] = None,
               bot_context: Optional[dict] = None) -> Optional[dict]:
        """Snapshot all features at a decision point. Returns the record
        (for logging only -- the bot must NOT act on it)."""
        now = now or datetime.now(ET)
        snap = self._build_snapshot(symbol, event, now, bot_context)
        self._write(snap, now)          # outside the lock: never blocks the stream thread
        return _clean(snap)

    def _build_snapshot(self, symbol: str, event: str, now: datetime,
                        bot_context: Optional[dict]) -> Dict[str, Any]:
        with self._lock:
            snap: Dict[str, Any] = {
                "type": "snapshot",
                "id": uuid.uuid4().hex[:12],
                "ts": now,
                "symbol": symbol,
                "event": str(event).upper(),
                "bot": bot_context or {},
            }
            st = self._states.get(symbol)
            # guard against stale prior-day state
            if st is None or not st.bars or st.date != _et_date(now):
                snap["error"] = "no_bars_today"
                return snap

            price = st.last_price if st.last_price is not None else st.bars[-1].close
            vol = self._volume_features(st, now)
            flow = self._flow_features(st, now)
            ctx = self._context_features(symbol, st, price)
            vas = self._vas(vol, flow)
            shadow = self._rules(vol, flow, ctx, vas)
            snap.update(price=price, volume=vol, flow=flow, context=ctx,
                        vas=vas, shadow=shadow)
            if self.cfg["track_outcomes"]:
                self._pending.setdefault(symbol, []).append({
                    "id": snap["id"], "symbol": symbol, "event": snap["event"],
                    "t0": now, "p0": price, "returns": {},
                    "mfe": None, "mae": None, "bars": 0,
                })
            return snap

    # ------------------------------------------------------------ features
    def _volume_features(self, st: _SymbolState, now: datetime) -> dict:
        c = self.cfg
        v = [b.volume for b in st.bars]
        n = len(v)
        L, mb, w = c["rvol_lookback_bars"], c["min_baseline_bars"], c["accel_window_bars"]
        out: Dict[str, Any] = {
            "bars": n, "last_bar_volume": v[-1] if v else None,
            "baseline_avg": None, "bar_rvol": None, "raw_accel": None,
            "smooth_accel": None, "accel_momentum": None,
            "rvol_3bar": None, "trend_3bar": "INSUFFICIENT",
            "projected_rvol": None,
        }
        if n < mb + 1:
            return out
        base = v[-(L + 1):-1]
        avg = sum(base) / len(base)
        out["baseline_avg"] = avg
        if avg <= 0:
            return out
        out["bar_rvol"] = v[-1] / avg

        # raw bar-over-bar accel, denominator floored so a near-empty prior
        # bar can't produce +2000% noise
        out["raw_accel"] = v[-1] / max(v[-2], c["raw_accel_floor_frac"] * avg) - 1.0

        # smoothed accel: last w bars vs the w bars before them
        if n >= 2 * w:
            out["smooth_accel"] = _ratio_minus_one(sum(v[-w:]), sum(v[-2 * w:-w]))
        if n >= 2 * w + 1 and out["smooth_accel"] is not None:
            prev = _ratio_minus_one(sum(v[-w - 1:-1]), sum(v[-2 * w - 1:-w - 1]))
            if prev is not None:
                out["accel_momentum"] = out["smooth_accel"] - prev

        # 3-bar participation trend (each bar vs its own trailing baseline)
        if n >= mb + 3:
            rv = []
            for k in (3, 2, 1):
                idx = n - k
                b = v[max(0, idx - L):idx]
                a = sum(b) / len(b)
                rv.append(v[idx] / a if a > 0 else None)
            if None not in rv:
                out["rvol_3bar"] = rv
                if rv[0] < rv[1] < rv[2]:
                    out["trend_3bar"] = "ACCELERATING"
                elif rv[0] > rv[1] > rv[2]:
                    out["trend_3bar"] = "DECELERATING"
                else:
                    out["trend_3bar"] = "MIXED"

        # in-progress bar, time-normalised (a 20 s partial bar compared raw
        # against a full bar would always look like deceleration)
        if st.cur_bar_start is not None:
            elapsed = (now - st.cur_bar_start).total_seconds()
            if c["min_projected_elapsed_s"] <= elapsed < c["bar_seconds"]:
                projected = st.cur_bar_volume * c["bar_seconds"] / elapsed
                out["projected_rvol"] = projected / avg
        return out

    def _flow_features(self, st: _SymbolState, now: datetime) -> dict:
        win = self.cfg["imbalance_window_s"]
        cutoff = now - timedelta(seconds=win)
        buy = sell = unk = 0.0
        count = 0
        for ts, _p, size, side in st.trades:
            if ts < cutoff or ts > now:
                continue
            count += 1
            if side > 0:
                buy += size
            elif side < 0:
                sell += size
            else:
                unk += size
        total = buy + sell
        out: Dict[str, Any] = {
            "window_s": win, "trades": count, "trade_rate": count / win,
            "buy_vol": buy, "sell_vol": sell, "unclassified_vol": unk,
            "imbalance": (buy - sell) / total if total > 0 else None,
            "bid": None, "ask": None, "spread_pct": None,
            "quote_size_imbalance": None, "quote_age_s": None,
        }
        if st.quote:
            qts, bid, ask, bsz, asz = st.quote
            out.update(bid=bid, ask=ask, quote_age_s=(now - qts).total_seconds())
            if bid > 0 and ask >= bid:
                out["spread_pct"] = (ask - bid) / ((ask + bid) / 2.0) * 100.0
            if bsz + asz > 0:
                out["quote_size_imbalance"] = (bsz - asz) / (bsz + asz)
        return out

    def _context_features(self, symbol: str, st: _SymbolState, price: float) -> dict:
        c = self.cfg
        vwap = st.cum_pv / st.cum_v if st.cum_v > 0 else None
        atr = st.atr if st.atr and st.atr > 0 else None
        ema_stack = None
        if st.ema9 is not None and st.ema20 is not None:
            ema_stack = price > st.ema9 > st.ema20
        spread_atr = None
        if st.quote and atr:
            spread_atr = (st.quote[2] - st.quote[1]) / atr

        k = c["rs_lookback_bars"]
        own = _change_pct(st.bars, k)
        rs: Dict[str, Optional[float]] = {}
        for bm in c["benchmark_symbols"]:
            if bm == symbol:
                continue
            bst = self._states.get(bm)
            bchg = _change_pct(bst.bars, k) if bst and bst.date == st.date else None
            rs[bm] = own - bchg if own is not None and bchg is not None else None

        return {
            "vwap": vwap, "price_vs_vwap_pct": _pct(price, vwap),
            "ema9": st.ema9, "ema20": st.ema20, "ema_stack": ema_stack,
            "ema_warm": len(st.bars) >= 20,
            "hod": st.hod, "dist_to_hod_pct": _pct(price, st.hod),
            "dist_to_hod_atr": (st.hod - price) / atr if atr and st.hod else None,
            "atr": atr, "spread_atr": spread_atr,
            "own_change_pct": own, "rel_strength_pct": rs,
        }

    # ------------------------------------------------------------ scoring
    def _vas(self, vol: dict, flow: dict) -> dict:
        c = self.cfg
        m, th = vol["accel_momentum"], c["momentum_threshold"]
        comps = {
            "rvol": _bucket(vol["bar_rvol"], c["rvol_buckets"]),
            "accel": _bucket(vol["smooth_accel"], c["accel_buckets"]),
            "momentum": None if m is None else (1 if m > th else -1 if m < -th else 0),
            "imbalance": _bucket(flow["imbalance"], c["imbalance_buckets"]),
        }
        # signed pressure: buyers/sellers (sign) x participation (magnitude).
        # NOT accel x imbalance: that turns "falling volume + sellers" positive.
        pressure = None
        if flow["imbalance"] is not None and vol["bar_rvol"] is not None:
            pressure = flow["imbalance"] * vol["bar_rvol"]

        lo = (min(s for _, s in c["rvol_buckets"]) + min(s for _, s in c["accel_buckets"])
              - 1 + min(s for _, s in c["imbalance_buckets"]))
        hi = (max(s for _, s in c["rvol_buckets"]) + max(s for _, s in c["accel_buckets"])
              + 1 + max(s for _, s in c["imbalance_buckets"]))
        missing = [k for k, s in comps.items() if s is None]
        raw = score = None
        if not missing:
            raw = sum(comps.values())
            score = round((raw - lo) / (hi - lo) * 10.0, 1)
        return {"raw": raw, "score": score, "components": comps,
                "missing": missing, "volume_pressure": pressure}

    def _rules(self, vol: dict, flow: dict, ctx: dict, vas: dict) -> dict:
        r = self.cfg["rules"]

        def chk(val, fn):
            return None if val is None else bool(fn(val))

        rs_vals = [x for x in ctx["rel_strength_pct"].values() if x is not None]
        trend = vol["trend_3bar"]
        rules = {
            "vas": chk(vas["score"], lambda x: x >= r["min_vas"]),
            "imbalance": chk(flow["imbalance"], lambda x: x >= r["min_imbalance"]),
            "bar_rvol": chk(vol["bar_rvol"], lambda x: x >= r["min_bar_rvol"]),
            "spread": chk(flow["spread_pct"], lambda x: x <= r["max_spread_pct"]),
            "rel_strength": chk(min(rs_vals) if rs_vals else None,
                                lambda x: x >= r["min_rel_strength_pct"]),
            "not_decelerating": None if trend == "INSUFFICIENT" else
                (trend != "DECELERATING" if r["block_decelerating"] else True),
        }
        failed = [k for k, ok in rules.items() if ok is False]
        missing = [k for k, ok in rules.items() if ok is None]
        decision = "FAIL" if failed else ("INCOMPLETE" if missing else "PASS")
        return {"decision": decision, "failed": failed, "missing": missing, "rules": rules}

    # ------------------------------------------------------------ outcomes
    def _update_outcomes(self, symbol: str, bar: Bar) -> None:
        pend = self._pending.get(symbol)
        if not pend:
            return
        horizons = self.cfg["outcome_horizons_s"]
        window = timedelta(seconds=max(horizons))
        bar_end = bar.ts + self._bar_td
        keep = []
        for o in pend:
            if _et_date(bar.ts) != _et_date(o["t0"]):
                self._emit_outcome(o, partial=True)
                continue
            if bar.ts < o["t0"]:  # bar began before the snapshot
                keep.append(o)
                continue
            if bar.ts - o["t0"] < window:
                hi = _pct(bar.high, o["p0"])
                lo = _pct(bar.low, o["p0"])
                o["mfe"] = hi if o["mfe"] is None else max(o["mfe"], hi)
                o["mae"] = lo if o["mae"] is None else min(o["mae"], lo)
                o["bars"] += 1
            for h in horizons:
                if str(h) not in o["returns"] and bar_end >= o["t0"] + timedelta(seconds=h):
                    o["returns"][str(h)] = _pct(bar.close, o["p0"])
            if len(o["returns"]) == len(horizons):
                self._emit_outcome(o, partial=False)
            else:
                keep.append(o)
        self._pending[symbol] = keep

    def _emit_outcome(self, o: dict, partial: bool) -> None:
        self._write({
            "type": "outcome", "id": o["id"], "symbol": o["symbol"],
            "event": o["event"], "t0": o["t0"], "p0": o["p0"],
            "returns_pct": o["returns"], "mfe_pct": o["mfe"], "mae_pct": o["mae"],
            "bars_observed": o["bars"], "partial": partial,
        }, o["t0"])

    @_safe
    def flush(self) -> None:
        """Call at end of day: writes partial outcomes for pending snapshots."""
        with self._lock:
            for pend in self._pending.values():
                for o in pend:
                    self._emit_outcome(o, partial=True)
            self._pending.clear()

    # ------------------------------------------------------------ io
    def _write(self, record: dict, ts: datetime) -> None:
        d = self.cfg["log_dir"]
        os.makedirs(d, exist_ok=True)
        path = os.path.join(d, f"shadow_entry_{ts.astimezone(ET):%Y-%m-%d}.jsonl")
        with open(path, "a", encoding="utf-8") as f:
            f.write(json.dumps(_clean(record), separators=(",", ":")) + "\n")


# ===========================================================================
# trade1 adapter (rules_api contract) -- [2026-09-27]
# ===========================================================================
# Ticks and bars arrive from the core through the module-level on_trade /
# on_quote / on_bar / seed_bars hooks (the stream's listener fan-out, which
# also delivers the benchmark symbols, config streaming.benchmark_symbols).
# evaluate() builds the engine's snapshot for the symbol right now and buys
# on a PASS verdict.
#
# Settings: config.json "entry_rules" (engine keys as in DEFAULT_CONFIG, plus):
#   max_entries_per_symbol  1     buys per stock per day
#   stop_lookback_bars      5     stop = lowest low of the last N 1-min bars
#   stop_buffer_cents       1.0     ... minus this
#   min_stop_pct            0.5   stop at least this % below the price
#   earliest_minute         0     no buys before this many minutes after the open
from rules_api import EntryDecision, MarketView  # noqa: E402

NAME = "entry_rules"
ADAPTER_DEFAULTS = {"max_entries_per_symbol": 1, "stop_lookback_bars": 5, "stop_buffer_cents": 1.0,
                    "min_stop_pct": 0.5, "earliest_minute": 0}
_engine: Optional[ShadowEntryRules] = None
_entries: Dict[Tuple[str, str], int] = {}


def _get_engine() -> ShadowEntryRules:
    global _engine
    if _engine is None:
        from config_loader import get_config
        cfg = dict(get_config().get("entry_rules", {}))
        cfg.update(enabled=True, track_outcomes=False)   # engine on whenever the module is loaded
        _engine = ShadowEntryRules(cfg)
    return _engine


def _to_bar(b: dict) -> Bar:
    return Bar(b["t"], b["o"], b["h"], b["l"], b["c"], b["v"])


def on_trade(symbol, ts, price, size):
    _get_engine().on_trade(symbol, ts, price, size)


def on_quote(symbol, ts, bid, ask, bid_size=0.0, ask_size=0.0):
    _get_engine().on_quote(symbol, ts, bid, ask, bid_size, ask_size)


def on_bar(symbol, bar):
    _get_engine().on_bar(symbol, _to_bar(bar))


def seed_bars(symbol, bars):
    _get_engine().seed_bars(symbol, [_to_bar(b) for b in bars])


def _indicators(snap: dict) -> dict:
    """The values worth keeping on the trade for later review."""
    v, f, c, s = snap.get("volume", {}), snap.get("flow", {}), snap.get("context", {}), snap.get("vas", {})
    return _clean({
        "verdict": snap.get("shadow", {}).get("decision"), "failed": snap.get("shadow", {}).get("failed"),
        "missing": snap.get("shadow", {}).get("missing"),
        "vas": s.get("score"), "vas_parts": s.get("components"), "volume_pressure": s.get("volume_pressure"),
        "bar_rvol": v.get("bar_rvol"), "projected_rvol": v.get("projected_rvol"), "smooth_accel": v.get("smooth_accel"),
        "accel_momentum": v.get("accel_momentum"), "trend_3bar": v.get("trend_3bar"),
        "imbalance": f.get("imbalance"), "trades_30s": f.get("trades"), "spread_pct": f.get("spread_pct"),
        "price_vs_vwap_pct": c.get("price_vs_vwap_pct"), "dist_to_hod_pct": c.get("dist_to_hod_pct"),
        "ema_stack": c.get("ema_stack"), "own_change_5bar_pct": c.get("own_change_pct"),
        "rel_strength_pct": c.get("rel_strength_pct"),
    })


def evaluate(view: MarketView, state: dict) -> EntryDecision:
    c = {**ADAPTER_DEFAULTS, **(view.cfg or {})}
    key = (view.now.astimezone(ET).date().isoformat(), view.symbol)
    if _entries.get(key, 0) >= c["max_entries_per_symbol"]:
        return EntryDecision(state="DONE", reason="max entries for this stock today")
    if view.minutes_since_open < c["earliest_minute"]:
        return EntryDecision(state="WAIT", reason="too early")
    snap = _get_engine()._build_snapshot(view.symbol, "EVAL", view.now, None)
    if snap.get("error"):
        return EntryDecision(state="WAIT", reason=snap["error"])
    ind = _indicators(snap)
    verdict = ind["verdict"]
    if verdict != "PASS":
        why = ("failed " + ",".join(ind["failed"])) if ind["failed"] else ("missing " + ",".join(ind["missing"] or []))
        return EntryDecision(state=verdict or "WAIT", reason=why, metrics={"vas": ind["vas"]})
    lows = [b["l"] for b in view.bars[-int(c["stop_lookback_bars"]):]]
    stop = min(min(lows) - c["stop_buffer_cents"] / 100.0, view.price * (1 - c["min_stop_pct"] / 100.0))
    _entries[key] = _entries.get(key, 0) + 1
    return EntryDecision(True, stop=round(stop, 4), state="BUY",
                         reason=f"entry_rules PASS (VAS {ind['vas']})", reasons=[f"VAS {ind['vas']}"],
                         plan={"indicators": ind}, metrics=ind)