feat(web): close #701 and #702 — the mark reconciliation pinned, the attestation chip, and a staleness verdict

keel/commands/balances.py

@@ -41,6 +41,7 @@
 from keel.commands.positions import PositionRow, gather_positions
 from keel.config import Config
 from keel.data.repository import Repository
+from keel.types import EquityReading
 
 
 @dataclass(frozen=True)
@@ -154,13 +155,80 @@ class BalancesReport:
 
     assets: tuple[AssetBalanceRow, ...]
 
+    #: Whether the newest reading has fallen behind this deployment's own cadence (#702).
+    #:
+    #: Decided HERE, not by a client counting seconds (Rule 2): "is 3 days old a problem" depends
+    #: on how often this deployment cycles, and that is a judgement. FALSE when there is no
+    #: reading at all -- a reading that does not exist is not an old one.
+    cash_stale: bool = False
+
+    #: The gap this deployment's recent readings actually arrive at, or `None` when there are
+    #: fewer than two to measure. `None` is not "fast": it is nothing to compare against, which is
+    #: why `_stale_window_sec` falls back to the configured interval rather than to zero.
+    observed_interval_sec: int | None = None
+
+    #: How far behind a reading may fall before `cash_stale`. Carried so the page can SAY the
+    #: threshold rather than leave a reader guessing why a stamp is or is not flagged.
+    stale_window_sec: int = 0
+
     @property
     def asset_count(self) -> int:
         """How many products this report holds. Derived, and held here rather than measured by a
         renderer: Rule 6e bans `len()` in `keel/web/payload.py`."""
         return len(self.assets)
 
 
+#: How many cycle intervals a reading may fall behind before the page calls it stale.
+#:
+#: Three, not two: a deployment that misses ONE cycle -- a restart, a slow venue, a laptop asleep
+#: for a beat -- is not a deployment that has stopped, and a badge that fires on a single skipped
+#: run is a badge an operator learns to ignore.
+STALE_INTERVALS = 3
+
+#: How many recent readings the observed cadence is measured over.
+STALE_SAMPLE = 8
+
+
+def _observed_interval_sec(readings: Sequence[EquityReading]) -> int | None:
+    """The deployment's own cadence, as the MEDIAN gap between its recent readings.
+
+    `None` when there are fewer than two readings, because one reading is no cadence -- and the
+    honest answer to "is this old?" with nothing to compare against is "not known", never "fine".
+
+    The MEDIAN rather than the mean or the last gap: a deployment that was off for a week has one
+    enormous gap in its history, and a mean would let that single outage widen the window
+    permanently. A median describes what this deployment normally does.
+    """
+    stamps = sorted(reading.ts for reading in readings)
+    if len(stamps) < 2:
+        return None
+    gaps = sorted(later - earlier for earlier, later in zip(stamps, stamps[1:], strict=False))
+    middle = len(gaps) // 2
+    if len(gaps) % 2:
+        return int(gaps[middle])
+    return int((gaps[middle - 1] + gaps[middle]) // 2)
+
+
+def _stale_window_sec(config: Config, observed: int | None) -> int:
+    """How far behind a reading may fall before it is stale.
+
+    **The LARGER of the configured interval and the deployment's own observed cadence, and the
+    second half is not a refinement -- it is what stops this badge crying wolf on the live
+    profile.**
+
+    `auto_trade.interval_sec` ships as 900, and the live deployment is driven by a wrapper that
+    runs the agent once per UTC DAY: the scheduler fires far more often and the wrapper decides.
+    Scaling the window off the config value alone would mark that deployment stale for about
+    twenty-three and a half hours out of every twenty-four, while it worked perfectly.
+
+    `agent._finest_granularity` records exactly this hazard for exactly this reason -- a slow
+    series "would spuriously flag a perfectly healthy feed as stale" -- and the answer here is the
+    same: judge a deployment against what it actually does.
+    """
+    configured = int(getattr(getattr(config, "auto_trade", None), "interval_sec", 0) or 0)
+    return STALE_INTERVALS * max(configured, observed or 0)
+
+
 def gather_balances(repo: Repository, config: Config, *, now_ts: int) -> BalancesReport:
     """Everything the account holds, from what a cycle wrote down. No broker, no network.
 
@@ -172,10 +240,15 @@ def gather_balances(repo: Repository, config: Config, *, now_ts: int) -> Balance
 
     reading = None
     balance = None
+    # Empty when no mode is stamped -- a deployment before its first cycle. The staleness read
+    # below runs over it either way and answers "nothing to judge", which is the honest answer.
+    recorded: list[EquityReading] = []
     if mode:
-        recorded = repo.get_equity_points(mode=mode, limit=1)
-        # `limit=1` keeps the MOST RECENT reading (`get_equity_points`' own contract), so this is
-        # one row off an index rather than the whole series read to take its last element.
+        # `STALE_SAMPLE`, not 1. The newest reading is still `recorded[-1]` (`get_equity_points`
+        # keeps the most recent and returns them oldest-first), and the ones behind it are what
+        # the observed cadence is measured over -- see `_stale_window_sec` for why a cadence read
+        # off the deployment matters more than the one the config declares.
+        recorded = repo.get_equity_points(mode=mode, limit=STALE_SAMPLE)
         reading = recorded[-1] if recorded else None
 
         # Same `limit=1`-keeps-the-newest contract, narrowed to the SETTLEMENT currency (#719) --
@@ -189,6 +262,8 @@ def gather_balances(repo: Repository, config: Config, *, now_ts: int) -> Balance
     # gate, so computing one would be three of every four candle reads plus a rules read and
     # a rule construction, per request, on a view the console re-polls every 15 seconds.
     positions = gather_positions(repo, config, now_ts=now_ts, with_readiness=False)
+    observed = _observed_interval_sec(recorded)
+    window = _stale_window_sec(config, observed)
     return BalancesReport(
         now_ts=now_ts,
         mode=mode,
@@ -198,6 +273,12 @@ def gather_balances(repo: Repository, config: Config, *, now_ts: int) -> Balance
         unrealized=None if reading is None else reading.unrealized,
         hwm=None if reading is None else reading.hwm,
         has_recorded_cash=reading is not None,
+        observed_interval_sec=observed,
+        stale_window_sec=window,
+        # A reading that does not exist is not an OLD one. `has_recorded_cash` says there is none,
+        # and a stale badge over a deployment that has never run would be a false alarm about a
+        # non-event.
+        cash_stale=reading is not None and (now_ts - reading.ts) > window,
         paper_cash=repo.get_state("paper_cash_usdc") if mode == "paper" else None,
         settled_cash=None if balance is None else balance.available,
         total_cash=None if balance is None else balance.total,

keel/commands/positions.py

@@ -40,10 +40,48 @@
 from keel_core.types import Granularity
 
 from keel import agent as agent_mod
+from keel.commands.doctor import ATTEST_WINDOW_APPROACHING_SEC as _DOCTOR_APPROACHING_SEC
 from keel.config import Config
 from keel.data import freshness as freshness_mod
 from keel.data.repository import Repository
 
+#: What a holding's asset attestation says, as a closed vocabulary (#701).
+#:
+#: FOUR words for four different facts, and the middle two are why this is not a boolean:
+#: `unattested` is the screen's own rejection, `expired` is a claim that has run out, `due` is one
+#: about to, and `attested` covers both "in date" and "no window recorded" -- which #718 made a
+#: legitimate state rather than a missing one.
+ATTESTATION_STATES: tuple[str, ...] = ("attested", "due", "expired", "unattested")
+
+#: How near a window has to be before the page says so. `doctor`'s own threshold, IMPORTED rather
+#: than restated: a page warning at one horizon beside a `keel doctor` warning at another would
+#: have an operator believing whichever they read last.
+ATTEST_APPROACHING_SEC = _DOCTOR_APPROACHING_SEC
+
+
+def _attestation_state(row: dict[str, Any] | None, *, now_ts: int) -> tuple[str, int | None]:
+    """`(state, due_ts)` for one asset's attestation row.
+
+    The window rules are `doctor._attestation_window_findings`', deliberately: `due <= now_ts` is
+    CLOSED (with `<`, a window landing exactly on the second falls into no group at all, which is
+    a false all-clear), and a NULL window is not judged at all.
+
+    **Reporting only, like doctor's.** `screen_asset` never reads `attest_due_ts`, so an expired
+    window does not veto anything -- #718 left that decision to a human. This page says what the
+    record holds and does not imply a block that is not there.
+    """
+    if row is None:
+        return "unattested", None
+    raw = row.get("attest_due_ts") if hasattr(row, "get") else row["attest_due_ts"]
+    if raw is None:
+        return "attested", None
+    due = int(raw)
+    if due <= now_ts:
+        return "expired", due
+    if due - now_ts <= ATTEST_APPROACHING_SEC:
+        return "due", due
+    return "attested", due
+
 
 @dataclass(frozen=True)
 class PositionRow:
@@ -103,6 +141,26 @@ class PositionRow:
     ready: bool
     ready_reason: str | None
 
+    #: What the operator has sworn about this holding's ASSET, and whether that claim is in date
+    #: (#701, on #718's recorded window). One of `ATTESTATION_STATES`.
+    #:
+    #: A holding is a claim about the world -- what the token is, what backs it -- and
+    #: `keel/compliance/screen.py`'s rule is that an asset nobody has classified is unknown and
+    #: unknown is a REJECTION. So absent reads `unattested`, never a quiet blank: a page rendering
+    #: it as fine would say the opposite of the gate that governs it.
+    #:
+    #: Distinct from `ready` above, which is the entry-gate verdict about DATA. The two disagree in
+    #: both directions and a reader needs to know which is which: an asset can be perfectly
+    #: attested with a cold series, or freshly priced with a lapsed claim.
+    #:
+    #: Defaulted and trailing, so every existing `PositionRow(...)` construction is untouched.
+    attestation: str = "unattested"
+
+    #: The recorded window's close, or `None` when the operator recorded none. NULL is NOT judged
+    #: -- `doctor._attestation_window_findings` states that convention and #718 made the column
+    #: optional deliberately, so no window is not an expired one.
+    attest_due_ts: int | None = None
+
 
 @dataclass(frozen=True)
 class PositionsReport:
@@ -286,6 +344,12 @@ def gather_positions(
     gates = _gate_granularities(repo, config) if with_readiness else {}
     fallback = _fallback_granularity(config) if with_readiness else None
 
+    # ONE read for the whole book, not one per tranche: this page re-polls every 15 seconds, and
+    # `get_asset_attestation` per row would be a query per row. Same shape as #707's
+    # `open_bracket_order_ids`.
+    attestations = {
+        str(row.get("asset") or "").upper(): row for row in repo.get_asset_attestations()
+    }
     marks: dict[str, tuple[Decimal | None, int | None]] = {}
     # Keyed on (product, gate granularity), NOT on product alone: one product can hold tranches
     # opened by rules on different timeframes, and a per-product cache would hand the second
@@ -307,16 +371,32 @@ def gather_positions(
         else:
             ready, ready_reason = False, None
         mark, mark_ts = marks[product_id]
-        rows.append(_row_from_dict(raw, mark, mark_ts, ready, ready_reason))
+        attestation, attest_due = _attestation_state(
+            attestations.get(_base_asset(product_id)), now_ts=now_ts
+        )
+        rows.append(
+            _row_from_dict(
+                raw, mark, mark_ts, ready, ready_reason, attestation, attest_due
+            )
+        )
     return PositionsReport(now_ts=now_ts, rows=tuple(rows))
 
 
+def _base_asset(product_id: str) -> str:
+    """`BTC-USD` -> `BTC`, the same split `compliance/screen.py` and `execution/guards.py` use to
+    key an asset attestation off a product. Upper-cased because `asset_attestations` is keyed on
+    the operator's own spelling and a page must not miss a claim over a letter case."""
+    return str(product_id).split("-")[0].upper()
+
+
 def _row_from_dict(
     raw: dict[str, Any],
     mark: Decimal | None,
     mark_ts: int | None,
     ready: bool,
     ready_reason: str | None,
+    attestation: str = "unattested",
+    attest_due_ts: int | None = None,
 ) -> PositionRow:
     """One repository dict, projected. Every judgement this report makes is made here, once, so
     no renderer has to make it twice."""
@@ -358,4 +438,6 @@ def _row_from_dict(
         realized_fees=raw.get("realized_fees") or Decimal("0"),
         ready=ready,
         ready_reason=ready_reason,
+        attestation=attestation,
+        attest_due_ts=attest_due_ts,
     )

keel/web/payload.py

@@ -775,6 +775,32 @@ def _bracket_field(position: OpenPositionStatus, mode: str) -> Field:
     return label("n/a", display="n/a -- paper resolves stop/target on candle touch", state=NEUTRAL)
 
 
+#: What each attestation state MEANS and how it reads (#701).
+#:
+#: `unattested` is UNKNOWN, not BAD: `screen_asset` rejects an unclassified asset, so the holding
+#: is already gated -- the page reports a missing claim rather than grading the operator for it.
+#: `expired` and `due` WARN and never FAIL, which is doctor's own choice for the same column and
+#: for the same reason: an expired window vetoes nothing (#718 left that decision to a human), and
+#: a FAIL would tell an operator the opposite.
+_ATTESTATION_CHIP: Mapping[str, tuple[str, str]] = {
+    "attested": ("attested", NEUTRAL),
+    "due": ("attestation window closing", WARN),
+    "expired": ("ATTESTATION EXPIRED", WARN),
+    "unattested": ("no attestation on record", UNKNOWN),
+}
+
+
+def _attestation_chip(row: PositionRow) -> Field:
+    """The chip, and the window behind it.
+
+    A `label` rather than a `flag`, because there are FOUR readings and two of them are not the
+    negation of the others: "no window recorded" and "window passed" are different facts about an
+    attestation that exists, and "never attested" is a different fact again.
+    """
+    display, state = _ATTESTATION_CHIP.get(row.attestation, ("unknown", UNKNOWN))
+    return label(row.attestation, display=display, state=state)
+
+
 def _position_payload(position: OpenPositionStatus, mode: str) -> dict[str, Any]:
     """One open position.
 
@@ -1985,6 +2011,12 @@ def _position_row_payload(row: PositionRow) -> dict[str, Any]:
         "realized_proceeds": money(row.realized_proceeds),
         "realized_fees": money(row.realized_fees),
         "freshness": _readiness_field(row.ready, row.ready_reason),
+        # #701. BESIDE the freshness verdict, never merged with it: one is a claim about the
+        # WORLD (what this token is, and whether the claim is in date) and the other is a
+        # claim about DATA. They disagree in both directions, and a reader needs to know
+        # which of the two is the reason a row is flagged.
+        "attestation": _attestation_chip(row),
+        "attest_due_at": moment(row.attest_due_ts),
     }
 
 
@@ -2030,6 +2062,55 @@ def _asset_balance_payload(row: AssetBalanceRow) -> dict[str, Any]:
     }
 
 
+def _human_span(seconds: int) -> str:
+    """A duration a reader can act on -- "3 days", "4 hours". No arithmetic in the client (Rule 2),
+    and no bare epoch difference on a page whose whole subject is when things were recorded."""
+    if seconds >= 172_800:
+        return f"{seconds // 86_400} days"
+    if seconds >= 7_200:
+        return f"{seconds // 3_600} hours"
+    if seconds >= 120:
+        return f"{seconds // 60} minutes"
+    return f"{seconds} seconds"
+
+
+def _staleness_payload(report: BalancesReport) -> Field:
+    """Whether the figures on this page describe now, and how confidently that can be said (#702).
+
+    THREE readings, and the middle one is why this is not a `flag`:
+
+    * **nothing recorded** -- a deployment before its first cycle. UNKNOWN: a reading that does
+      not exist is not an old one, and a stale badge here would be a false alarm about a
+      non-event.
+    * **behind this deployment's own cadence** -- WARN, and it names the age. A venue outage holds
+      the last row for as long as it lasts, so this is the difference between a current figure and
+      one that merely looks current.
+    * **current** -- NEUTRAL and never GOOD. A fresh reading is the ordinary state, and grading it
+      green would make the absence of green read as a fault on every page that has just started.
+
+    The JUDGEMENT is the report's (`cash_stale`), made against the cadence this deployment
+    actually cycles at rather than the one its config declares -- see `_stale_window_sec` for the
+    live profile that would otherwise be marked stale twenty-three hours a day.
+    """
+    if not report.has_recorded_cash:
+        return label(
+            "unrecorded",
+            display="no cycle has recorded a reading yet",
+            state=UNKNOWN,
+        )
+    age = report.now_ts - (report.cash_as_of or report.now_ts)
+    if report.cash_stale:
+        return label(
+            "stale",
+            display=(
+                f"STALE — the last reading is {_human_span(age)} old, past this deployment's "
+                f"own {_human_span(report.stale_window_sec)} window"
+            ),
+            state=WARN,
+        )
+    return label("current", display=f"recorded {_human_span(age)} ago", state=NEUTRAL)
+
+
 def balances_payload(report: BalancesReport) -> dict[str, Any]:
     """`gather_balances`'s `BalancesReport`, as JSON (#702).
 
@@ -2089,6 +2170,9 @@ def balances_payload(report: BalancesReport) -> dict[str, Any]:
         ),
         "asset_count": count(report.asset_count),
         "assets": [_asset_balance_payload(row) for row in report.assets],
+        # #702. The judgement, not the subtraction: "is 3 days old a problem" depends on how
+        # often this deployment cycles, and Rule 2 keeps that in Python.
+        "freshness": _staleness_payload(report),
     }