modules/decay_tracker.py

"""
modules/decay_tracker.py — measure how fast an edge dies between detection
and send, and gate on it (v1.0, 2026-07-29).

Operator, after 1,093 scan cycles and 0 signals: "Where's the signal we are
at Frankfurt zone". And, correctly, from their own analysis:

    "Log (quote_profit, requote_profit, elapsed_ms) on every attempt, hit or
     miss — right now you're tuning blind. That dataset tells you real
     decay, not guessed decay."

    Catch condition:
        Net_at_detection ≥ MIN_PROFIT_FLOOR + Decay_Rate × T_pipeline

    "The lever that actually catches more chances is shrinking T_pipeline,
     not lowering MIN_PROFIT_FLOOR. Lowering the floor just admits more
     trades that this same equation says you'll lose to decay anyway."

That is exactly right, and it is the piece this codebase never had. Worth
being precise about why it matters here.

────────────────────────────────────────────────────────────────────────────
THE PATTERN IN THIS DEPLOYMENT'S OWN HISTORY
────────────────────────────────────────────────────────────────────────────
Read solana_arb.py's changelog and the same event recurs under four
different diagnoses:

  v1.42  a real +$1.51 signal's safety re-quote queued behind the rest of a
         16-route cycle; ~20s later the edge was -$1.04 and it aborted.
  v1.47  the scanner quoted leg 2 for the full leg-1 output while the
         executor quoted it minus a haircut — signals that could never
         reproduce at send time.
  v1.16  a genuine +1.0 bps / +$1.95 signal died to a 429 moments after the
         scan loop tripped the shared cooldown.
  v1.50  a re-quote netting +$0.01 passed the margin gate and landed as a
         real loss once network costs were counted.

Every one was found by hand, from logs, after the fact. Every one is the
same shape: **the number that justified the trade was measured at time T,
and the trade happened at time T + Δ.** Nothing in the bot has ever
measured Δ, or what it costs.

So the floor was tuned blind. $10 → $0.20 → $0.03 across this changelog,
each move a guess at "let more through", when the binding constraint was
never the height of the bar — it was that the edge was already gone by the
time anything jumped.

────────────────────────────────────────────────────────────────────────────
WHAT THIS MODULE DOES
────────────────────────────────────────────────────────────────────────────
1. RECORDS every real execution attempt as one sample:

       quote_net_usd     what the scanner saw, at detection
       requote_net_usd   what the mandatory pre-send re-quote saw
       elapsed_ms        detection → re-quote, wall clock
       route, outcome

   These three numbers together are the dataset that has never existed.
   They are appended to their own CSV (separate from trade_journal.csv,
   which is one row per EVALUATION — this is one row per real ATTEMPT, a
   thousand times rarer and a different unit of analysis).

2. ESTIMATES the decay rate in USD per second, per route and globally, as
   the median of (quote_net − requote_net) / elapsed_secs over recent
   samples. Median, not mean: a single 429-delayed attempt with a 40s
   elapsed would otherwise dominate the estimate for hours.

3. GATES. `required_floor()` returns

       MIN_PROFIT_FLOOR + decay_rate × T_pipeline_p90

   so a signal must clear not just the floor, but the floor plus what the
   edge is expected to lose while the pipeline runs. A signal that cannot
   clear it was never going to land profitably — attempting it burns
   Jupiter budget and, at worst, real fees.

────────────────────────────────────────────────────────────────────────────
THE HONEST LIMITS
────────────────────────────────────────────────────────────────────────────
Decay is not one number. It is a mixture of at least three processes: other
searchers taking the same edge (fast, adversarial), ordinary price drift
(slow, symmetric), and our own pipeline stalling (bursty, self-inflicted).
A median lumps them together. That is still enormously better than the zero
numbers available before, and the per-route breakdown separates the worst
of it — but it should not be read as a physical constant.

The gate is also deliberately ASYMMETRIC. It only ever RAISES the bar, and
only once there are DECAY_MIN_SAMPLES real attempts to derive it from.
With no data it returns the static floor unchanged — this module cannot
make the bot trade more, only stop it attempting trades the data says are
already lost. `T_pipeline` uses the p90 of observed elapsed times rather
than the median, because the trade that matters is the slow one.

Set DECAY_GATE_ENABLED=false to record without gating — recommended for
the first day, so the numbers can be read (`/decay`) before they act.
"""

from __future__ import annotations

import csv
import logging
import os
import threading
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Optional

logger = logging.getLogger(__name__)


def _env_float(name: str, default: float) -> float:
    try:
        raw = os.getenv(name, "").strip()
        return float(raw) if raw else default
    except (TypeError, ValueError):
        return default


def _env_int(name: str, default: int) -> int:
    try:
        raw = os.getenv(name, "").strip()
        return int(float(raw)) if raw else default
    except (TypeError, ValueError):
        return default


def _env_bool(name: str, default: bool) -> bool:
    raw = os.getenv(name, "").strip().lower()
    if not raw:
        return default
    return raw in ("1", "true", "yes", "on")


# Recording is always on — it is three floats per real attempt and it is the
# only way this deployment will ever stop guessing.
RECORD_ENABLED = _env_bool("DECAY_RECORD_ENABLED", True)
# Gating ships OFF. Read the numbers for a day first; a gate derived from a
# handful of samples is a guess with extra steps.
GATE_ENABLED = _env_bool("DECAY_GATE_ENABLED", False)
_MIN_SAMPLES = max(5, _env_int("DECAY_MIN_SAMPLES", 20))
_WINDOW = max(_MIN_SAMPLES, _env_int("DECAY_WINDOW_SAMPLES", 200))

# Age ceiling on a decay sample. A measurement of how much edge died in a
# 2-second pipeline says nothing about a 0.2-second one, and keeping it
# lets a rebuilt pipeline stay priced at its old cost forever — see
# _recent() for the loop that produced.
_MAX_AGE_SECS = max(3600.0, float(_env_int("DECAY_MAX_AGE_SECS", 86400)))
# How many of the newest samples count as "recent" for the trend split.
# 20 is small enough to move within an hour of real attempts and large
# enough that one outlier cannot define it.
_RECENT_N = max(5, _env_int("DECAY_RECENT_SAMPLES", 20))
# Cap the gate's contribution so a pathological sample can't lock trading
# out entirely — same reasoning as dynamic_floor's own multiplier cap.
_MAX_ADDER_USD = _env_float("DECAY_MAX_ADDER_USD", 5.0)
_PATH = os.getenv("DECAY_LOG_PATH", "").strip() or "data/decay_log.csv"

_COLUMNS = [
    "ts_iso", "route", "quote_net_usd", "requote_net_usd", "elapsed_ms",
    "decay_usd", "decay_usd_per_sec", "outcome", "detail",
]


@dataclass
class _Sample:
    route: str
    quote_net_usd: float
    requote_net_usd: float
    elapsed_ms: float
    at: float = field(default_factory=time.time)

    @property
    def decay_usd(self) -> float:
        return self.quote_net_usd - self.requote_net_usd

    @property
    def decay_per_sec(self) -> float:
        secs = self.elapsed_ms / 1000.0
        return self.decay_usd / secs if secs > 0.05 else 0.0


def _percentile(values: list[float], q: float) -> float:
    if not values:
        return 0.0
    ordered = sorted(values)
    idx = min(len(ordered) - 1, max(0, int(round(q * (len(ordered) - 1)))))
    return ordered[idx]


def _median(values: list[float]) -> float:
    return _percentile(values, 0.5)


class DecayTracker:
    def __init__(self) -> None:
        self._samples: list[_Sample] = []
        self._lock = threading.Lock()
        self._path = Path(_PATH).expanduser()
        self._ready = False
        if RECORD_ENABLED:
            try:
                self._path.parent.mkdir(parents=True, exist_ok=True)
                if not self._path.exists() or self._path.stat().st_size == 0:
                    with self._path.open("w", newline="", encoding="utf-8") as fh:
                        csv.writer(fh).writerow(_COLUMNS)
                self._ready = True
                self._hydrate()
                logger.info(
                    "[DecayTracker] recording attempts to %s (%d prior sample(s) "
                    "loaded)", self._path, len(self._samples),
                )
            except Exception as exc:  # noqa: BLE001 — telemetry never blocks trading
                logger.warning("[DecayTracker] could not open %s (%s)", self._path, exc)

    def _hydrate(self) -> None:
        """Reload recent samples from the CSV at startup (v1.1).

        Caught in review on PR #132: samples lived in memory only, so every
        restart reset the estimator to zero. That is fatal for THIS
        deployment specifically — the operator restarts on every config
        change, often several times an hour, and real execution attempts
        are rare. The window would have reset faster than it could ever
        fill, so `/decay` would sit at "0/20 attempts" forever and the gate
        could never engage no matter how long the bot ran.

        Reads only the tail, tolerates a truncated first line, and skips
        any row it cannot parse. A corrupt or missing file means "start
        empty" — never a crash on startup.
        """
        try:
            size = self._path.stat().st_size
            with self._path.open("rb") as fh:
                # ~200 bytes/row; read generously and slice after parsing.
                budget = _WINDOW * 400
                if size > budget:
                    fh.seek(size - budget)
                    fh.readline()  # drop the partial row the seek landed in
                blob = fh.read().decode("utf-8", errors="replace")
        except OSError as exc:
            logger.debug("[DecayTracker] no prior samples to load: %s", exc)
            return

        loaded: list[_Sample] = []
        for row in csv.reader(blob.splitlines()):
            if len(row) != len(_COLUMNS) or row[0] == "ts_iso":
                continue
            try:
                # v1.17 — CARRY THE ROW'S OWN TIMESTAMP.
                #
                # `at` defaults to time.time(), so loading rebuilt every
                # historical sample as if it had just been measured. That
                # alone would have defeated the age ceiling added in this
                # version: the operator restarts several times an hour, and
                # each restart would have stamped five-day-old measurements
                # as fresh and handed them straight back to the gate.
                #
                # Column 0 is the ISO timestamp the row was written with.
                # A row whose timestamp will not parse is dropped rather
                # than dated to now — an undateable sample cannot be aged
                # out, and one that cannot be aged out is exactly what this
                # is here to prevent.
                at = datetime.fromisoformat(
                    row[0].replace("Z", "+00:00")).timestamp()
                loaded.append(_Sample(
                    route=row[1], quote_net_usd=float(row[2]),
                    requote_net_usd=float(row[3]), elapsed_ms=float(row[4]),
                    at=at,
                ))
            except (TypeError, ValueError):
                continue
        if loaded:
            self._samples = loaded[-_WINDOW:]

    # ── Recording ────────────────────────────────────────────────────────
    def record(
        self, route: str, quote_net_usd: float, requote_net_usd: float,
        elapsed_ms: float, outcome: str = "", detail: str = "",
    ) -> None:
        """One real execution attempt. Never raises."""
        try:
            sample = _Sample(
                route=route, quote_net_usd=float(quote_net_usd),
                requote_net_usd=float(requote_net_usd), elapsed_ms=float(elapsed_ms),
            )
        except (TypeError, ValueError):
            return
        with self._lock:
            self._samples.append(sample)
            if len(self._samples) > _WINDOW * 2:
                del self._samples[: len(self._samples) - _WINDOW]
        logger.info(
            "[DecayTracker] %s — quote $%+.4f -> re-quote $%+.4f after %.0fms "
            "(decay $%.4f, $%.4f/s) — %s",
            route, sample.quote_net_usd, sample.requote_net_usd, sample.elapsed_ms,
            sample.decay_usd, sample.decay_per_sec, outcome or "?",
        )
        if not self._ready:
            return
        try:
            row = [
                datetime.now(timezone.utc).isoformat(timespec="seconds"), route,
                round(sample.quote_net_usd, 6), round(sample.requote_net_usd, 6),
                round(sample.elapsed_ms, 1), round(sample.decay_usd, 6),
                round(sample.decay_per_sec, 6), outcome, (detail or "")[:200],
            ]
            with self._lock, self._path.open("a", newline="", encoding="utf-8") as fh:
                csv.writer(fh).writerow(row)
        except Exception as exc:  # noqa: BLE001
            logger.debug("[DecayTracker] row skipped: %s", exc)

    # ── Estimation ───────────────────────────────────────────────────────
    def _recent(self, route: Optional[str] = None) -> list[_Sample]:
        """The window, bounded by AGE as well as by count (v1.17).

        ────────────────────────────────────────────────────────────────
        THE FOREVER-LOOP THIS CLOSES
        ────────────────────────────────────────────────────────────────
        The window was 200 samples and nothing else. `_Sample.at` has
        existed since v1.0 and was never read, so a measurement taken on a
        pipeline that has since been rebuilt still set today's floor.

        On the operator's box that became self-sustaining:

            old 2.05s pipeline  → decay measured at $0.92/s
            → gate charges $2.566 of headroom
            → a configured $0.05 floor becomes $2.62
            → nothing clears $2.62, so nothing sends
            → no new attempt is ever recorded
            → the 20 ancient samples remain the whole window, forever

        Their /pipeline now reports 1.19s over 3 attempts while /decay
        still reports 2.05s over 20 — the same shape as the flash-fee
        loop fixed earlier: a stale safety number pricing itself into the
        gate that prevents its own replacement.

        Samples older than DECAY_MAX_AGE_SECS (24h) are dropped. If that
        leaves too few, the caller sees an under-sampled window and the
        gate stands down — which is correct: no recent evidence is a
        reason to stop charging for decay, not a reason to keep charging
        yesterday's rate.
        """
        cutoff = time.time() - _MAX_AGE_SECS
        with self._lock:
            samples = [s for s in self._samples[-_WINDOW:]
                       if getattr(s, "at", 0.0) >= cutoff]
        if route:
            samples = [s for s in samples if s.route == route]
        return samples

    def decay_usd_per_sec(self, route: Optional[str] = None) -> Optional[float]:
        """Median decay rate, or None without a real sample.

        Route-specific when there is enough of it, global otherwise: a
        per-route estimate off three samples is noise wearing a label.
        Negative medians are clamped to 0 — an edge that reliably GREW
        between quote and re-quote is a sampling artefact, and betting on
        it would be exactly the wrong direction to be wrong in.
        """
        samples = self._recent(route)
        if route and len(samples) < _MIN_SAMPLES:
            samples = self._recent(None)
        if len(samples) < _MIN_SAMPLES:
            return None
        rates = [s.decay_per_sec for s in samples if s.elapsed_ms > 50]
        if not rates:
            return None
        return max(0.0, _median(rates))

    def pipeline_secs_p90(self) -> Optional[float]:
        samples = self._recent(None)
        if len(samples) < _MIN_SAMPLES:
            return None
        return _percentile([s.elapsed_ms for s in samples], 0.90) / 1000.0

    def _gate_samples(self, route: Optional[str] = None) -> list[_Sample]:
        """Samples the gate derives its adder from — RECENT first (v1.3).

        The long window is right for describing history and wrong for
        setting a bar, because a structural change to the pipeline takes
        hours to work through 200 samples. Prefer the newest _RECENT_N when
        that slice is itself large enough to be an estimate rather than an
        anecdote; fall back to the full window otherwise.
        """
        samples = self._recent(route)
        if route and len(samples) < _MIN_SAMPLES:
            samples = self._recent(None)
        if len(samples) >= _RECENT_N >= _MIN_SAMPLES:
            return samples[-_RECENT_N:]
        return samples

    def decay_usd_p90(self, route: Optional[str] = None) -> Optional[float]:
        """p90 of the dollars actually lost between quote and re-quote.

        This is the number the gate wants, and it is measured directly.
        Clamped at 0: a p90 decay below zero would mean the edge usually
        GREW, which is not something to hand back as a discount.
        """
        samples = self._gate_samples(route)
        if len(samples) < _MIN_SAMPLES:
            return None
        return max(0.0, _percentile([s.decay_usd for s in samples], 0.90))

    def required_floor(
        self, static_floor_usd: float, route: Optional[str] = None,
    ) -> tuple[float, str]:
        """(floor_to_enforce, human explanation).

        Implements the operator's catch condition:
            Net_at_detection ≥ MIN_PROFIT_FLOOR + (edge lost during the pipeline)

        v1.3 — the second term is now MEASURED, not reconstructed.

        It used to be `decay_rate_$/s x pipeline_p90_secs`, and that product
        broke the moment the pipeline got faster. Two independent reasons,
        both visible in this deployment's own numbers on 2026-07-30:

          THE RATE INFLATES AS THE PIPELINE SHRINKS. decay_per_sec is
          decay_usd / elapsed_secs, and most of decay_usd is quote noise
          that does NOT shrink with elapsed time — only the adversarial and
          drift components do. So the same ~$1.6 of quote wobble reads as
          $0.13/s over a 12s pipeline and $1.07/s over a 1.5s one. The
          measured rate DOUBLED here at the exact moment the pipeline got
          eight times faster. That is division, not the market.

          THE TWO FACTORS CAME FROM DIFFERENT WORLDS. The rate is dominated
          by new fast samples while the p90 elapsed still lags on 200 rows
          of the old slow pipeline. Multiplying them produced $15.69 — an
          adder describing a pipeline that no longer exists, saturating the
          $5.00 cap and reporting the cap as if it were a finding. On edges
          worth $2-5, an enabled gate would have refused literally every
          trade, and the /decay output would have explained the refusal with
          a number that was never observed.

        The dataset already contains decay_usd per attempt. Reconstructing
        it from a rate and a duration only makes sense when extrapolating to
        a duration you have not observed — and here we have observed it. So
        take the p90 of the dollars directly. It cannot decouple from the
        pipeline, because it IS what the pipeline cost.

        The rate is still computed and still reported: "how much does a
        second cost me" is exactly the question that says whether more
        latency work pays. It just no longer sets the bar.
        """
        static = max(0.0, float(static_floor_usd))
        if not GATE_ENABLED:
            return static, "decay gate off (DECAY_GATE_ENABLED=false) — recording only"
        observed = self.decay_usd_p90(route)
        if observed is None:
            return static, (
                f"fewer than {_MIN_SAMPLES} real attempts recorded — no decay "
                f"estimate yet, static floor unchanged"
            )
        adder = min(_MAX_ADDER_USD, observed)
        n = len(self._gate_samples(route))
        return static + adder, (
            f"observed decay p90 ${observed:.4f} over the last {n} attempt(s) "
            f"= ${adder:.4f} on top of ${static:.2f}"
            + (f" (CAPPED at ${_MAX_ADDER_USD:.2f})" if adder >= _MAX_ADDER_USD else "")
        )

    # ── Introspection (/decay) ───────────────────────────────────────────
    def status(self) -> dict[str, Any]:
        samples = self._recent(None)
        by_route: dict[str, list[_Sample]] = {}
        for s in samples:
            by_route.setdefault(s.route, []).append(s)
        rate = self.decay_usd_per_sec()
        pipeline = self.pipeline_secs_p90()

        # v1.2 — the RECENT slice, reported next to the full window.
        #
        # The window is 200 samples and hydrates from disk, so a change to
        # the pipeline is invisible here for hours: a p90 over a window
        # still dominated by pre-change samples reports the OLD pipeline
        # and looks like the fix did nothing. It did that immediately after
        # the execute-on-detection change — 14.64s became 15.63s, which
        # reads as a regression and is really just the same old samples
        # plus one.
        #
        # A metric that cannot show an improvement for hours is a metric
        # nobody can act on. The last _RECENT_N samples are now summarised
        # separately, so the effect of a change is visible within a handful
        # of attempts and the long window keeps doing its job for the gate.
        recent = samples[-_RECENT_N:] if samples else []
        older = samples[:-_RECENT_N] if len(samples) > _RECENT_N else []

        def _slice(rows: list[_Sample]) -> dict[str, Any]:
            if not rows:
                return {"samples": 0, "median_secs": None, "median_decay_usd": None}
            return {
                "samples": len(rows),
                "median_secs": round(_median([s.elapsed_ms for s in rows]) / 1000.0, 3),
                "median_decay_usd": round(_median([s.decay_usd for s in rows]), 4),
            }

        recent_stats, older_stats = _slice(recent), _slice(older)
        trend = None
        if recent_stats["median_secs"] is not None and older_stats["median_secs"]:
            trend = round(
                recent_stats["median_secs"] - older_stats["median_secs"], 3,
            )

        # v1.3 — is the "decay" directional, or is it noise?
        #
        # Decay is supposed to be adversarial: other searchers take the edge,
        # so it should shrink far more often than it grows. If instead the
        # edge grows on roughly half the attempts, then what is being
        # measured is not decay at all — it is the ordinary variance between
        # two Jupiter quotes taken moments apart, and the median of it is a
        # number with no mechanism behind it.
        #
        # That distinction decides where the next hour of work goes. Skewed
        # means latency work pays. Symmetric means latency work is finished
        # and the constraint is that the edges are inside the noise band.
        # Without this split both look identical in the median.
        grew = sum(1 for s in samples if s.decay_usd < 0)
        gate_rows = self._gate_samples(None)
        observed_p90 = self.decay_usd_p90()

        return {
            "recording": self._ready,
            "gate_enabled": GATE_ENABLED,
            "path": str(self._path),
            "samples": len(samples),
            "min_samples": _MIN_SAMPLES,
            # Newest first in the reading order that matters: "is it better
            # NOW" before "what is the long-run average".
            "recent": recent_stats,
            "earlier": older_stats,
            "pipeline_trend_secs": trend,
            "recent_n": _RECENT_N,
            "decay_usd_per_sec": None if rate is None else round(rate, 5),
            # What the gate actually charges: dollars observed, not a rate
            # multiplied by a duration from a different window.
            "observed_decay_p90_usd": (
                None if observed_p90 is None else round(observed_p90, 4)
            ),
            "gate_window": len(gate_rows),
            "gate_capped": (
                observed_p90 is not None and observed_p90 >= _MAX_ADDER_USD
            ),
            "max_adder_usd": _MAX_ADDER_USD,
            # The noise check. `grew` counts attempts where the edge was
            # BIGGER at send time than at detection. Adversarial decay
            # cannot do that often; symmetric quote noise does it half the
            # time. Near 50% means the median "decay" has no mechanism
            # behind it and latency work will not move it.
            "grew_count": grew,
            "grew_share": round(grew / len(samples), 3) if samples else None,
            # v1.4 — the plain median decay across all attempts, in dollars.
            # Everything above reports a rate, a p90 or a per-route figure;
            # none of them is the like-for-like number modules/decay_control
            # needs to set the treatment arm against its control. Comparing
            # a p90 to a median would make the control look flat no matter
            # what it did.
            "median_decay_usd": (
                round(_median([s.decay_usd for s in samples]), 5) if samples else None
            ),
            "pipeline_p90_secs": None if pipeline is None else round(pipeline, 3),
            "pipeline_median_secs": (
                round(_median([s.elapsed_ms for s in samples]) / 1000.0, 3)
                if samples else None
            ),
            # v1.3 — the adder the gate WOULD apply, from the same code path
            # the gate uses. It used to be recomputed here as rate x p90,
            # which meant /decay could report a number the gate would never
            # charge. One source of truth.
            "implied_adder_usd": (
                None if observed_p90 is None
                else round(min(_MAX_ADDER_USD, observed_p90), 4)
            ),
            "routes": [
                {
                    "route": route,
                    "attempts": len(rows),
                    "median_decay_usd": round(_median([s.decay_usd for s in rows]), 4),
                    "median_elapsed_ms": round(_median([s.elapsed_ms for s in rows])),
                    "survived": sum(1 for s in rows if s.requote_net_usd > 0),
                }
                for route, rows in sorted(
                    by_route.items(), key=lambda kv: -len(kv[1]),
                )[:8]
            ],
        }


_tracker: Optional[DecayTracker] = None


def get_tracker() -> DecayTracker:
    global _tracker
    if _tracker is None:
        _tracker = DecayTracker()
    return _tracker