diff --git a/monitoring.yaml b/monitoring.yaml
index cf3ef603..d5e93dd1 100644
--- a/monitoring.yaml
+++ b/monitoring.yaml
@@ -45,6 +45,10 @@ protocols:
         description: "Alert-once when ProtocolConfig IS_PAUSED flips true"
       - name: "Borrower Default Watch"
         description: "Envio-backed MorphoCredit borrower watch; MEDIUM alerts when unpaid obligations become delinquent after grace or reach default"
+      - name: "Proof of Solvency"
+        description: "Accountable collateral ratio <95% (CRITICAL, after 2 consecutive runs) or <99% (HIGH)"
+      - name: "Proof of Solvency Freshness"
+        description: "MEDIUM when a short-cadence report/source misses 2 periods or a >1h cadence misses 1; HIGH after an unavailable feed exhausts retries"
       - name: "Timelock"
         description: "CallScheduled events from 24h and 7-day TimelockControllers via Envio"
 
diff --git a/protocols/3jane/README.md b/protocols/3jane/README.md
index a3120806..4fbbd985 100644
--- a/protocols/3jane/README.md
+++ b/protocols/3jane/README.md
@@ -15,6 +15,7 @@
 - **Nominal sUSD3 Backing Floor:** `ProtocolConfig.config(keccak256("SUSD3_NOMINAL_BACKING_FLOOR"))` vs cached prior. Alerts on any change (governance lever). Separate alert-once when the floor exceeds sUSD3's USD3 holdings valued in USDC — sUSD3 redemptions can be blocked while floor > backing.
 - **Protocol Pause:** `ProtocolConfig.config(keccak256("IS_PAUSED"))`. Alert-once on transition to true. Distinct from per-vault `isShutdown()` — pauses the underlying credit market.
 - **Borrower Default Watch:** optional Envio-backed borrower default risk feed. The Envio indexer maintains `ThreeJaneBorrowerMarket` rows from MorphoCredit events, and the monitor computes the current delinquent/default status at runtime. Alerts are **MEDIUM only** and deduped per borrower/cycle/default milestone.
+- **Proof of Solvency:** [Accountable](https://accountable.3jane.xyz/) collateral ratio (reserves / liabilities). Alerts **CRITICAL below 95%** and **HIGH below 99%**, plus freshness and availability alerts. See [Proof of Solvency](#proof-of-solvency) below.
 
 ## Key Contracts
 
@@ -43,6 +44,10 @@
 | Nominal floor breach | Floor > sUSD3 backing valued in USDC (alert-once) | MEDIUM |
 | Protocol paused | `IS_PAUSED` transitions to true (alert-once) | CRITICAL |
 | Borrower delinquent/default watch | New milestone: delinquent, ≤14d, ≤7d, ≤3d, ≤1d, default | MEDIUM |
+| Accountable collateral ratio | < 95% for 2 consecutive runs (band transition) | CRITICAL |
+| Accountable collateral ratio | < 99% (band transition) | HIGH |
+| Accountable feed stale | Short cadence >2 periods; long cadence >1 period (alert-once) | MEDIUM |
+| Accountable feed unavailable | One exhausted retrieval cycle (alert-once until recovery) | HIGH |
 | Monitoring run failure | Uncaught exception in `main()` | LOW |
 
 ## Borrower default watch
@@ -80,6 +85,40 @@ The indexer should populate/update that entity from `SetCreditLine`, `Borrow`, `
 
 The current countdown and alert bucket are intentionally computed in this monitoring script, not in Envio, because they depend on wall-clock time. Grace and delinquency windows default to 7 days and 23 days respectively in the indexer, and can be overridden there with `THREE_JANE_GRACE_PERIOD_SECONDS` and `THREE_JANE_DELINQUENCY_PERIOD_SECONDS`.
 
+## Proof of Solvency
+
+[Accountable](https://docs.accountable.capital/accountable-documentation/proof-of-solvency) publishes a TEE-attested Proof of Solvency dashboard for 3Jane (feed id `100000026`). The human-readable UI is at `https://accountable.3jane.xyz/` (override with `THREE_JANE_ACCOUNTABLE_MESSAGE_URL`); the JSON report is at `https://accountable.3jane.xyz/dashboard` (override with `THREE_JANE_ACCOUNTABLE_URL`). No API key is required.
+
+The client lives in [`utils/accountable.py`](../../utils/accountable.py) and is keyed by data feed id (DFID), so other Accountable feeds can be added without a rewrite. The request is URL/type-based and neither sends nor echoes the DFID, so feed identity is bound explicitly in config.
+
+### Ratio is recomputed, not read
+
+The API rounds `collateralization` to six decimals. Near the alert boundary that is a missed-critical-alert path: a true ratio of `0.9499996` would present as `0.95` and pass a `< 0.95` test. The monitor therefore computes the ratio from `total_reserves / total_supply` at full precision and uses the reported field only as a consistency cross-check (tolerance ≥1e-6, since the server's own rounding sets the floor).
+
+`net` and `collateralization` are defined against *liabilities*, which equal `total_supply` only for a USD-pegged feed. The client asserts `total_supply.fx == 1` when the field is present. The live response currently omits it, so that path independently derives liabilities from `total_reserves - net` and requires them to match raw supply; a non-pegged feed still fails loudly instead of silently comparing against the wrong denominator.
+
+### Freshness is per source, not global
+
+A fresh aggregate timestamp does not prove every input is fresh, and this matters more than usual here: `reserves_split` is essentially all "Morpho Credit", of which the bulk is off-chain loan receivables priced by manually uploaded document reports. Those routinely run past their declared cadence.
+
+The aggregate report and each required source use their declared cadence. Cadences of one hour or less get one missed-period allowance and become stale after two periods; longer cadences become stale as soon as the first expected update is late. The aggregate cadence comes from `reserves.interval`; source cadences come from each source's `frequency`. This means `15 MIN` becomes stale after 30 minutes, hourly after 2 hours, daily after 24 hours, and weekly after 7 days. A source whose `lastUpdated` is in the future is treated as unusable rather than clamped to "fresh", which would defeat the check. Unknown additional sources with an unrecognised cadence are skipped rather than flagged, so a schema addition on Accountable's side cannot spuriously page us.
+
+The 3Jane dashboard UI declares `Slope - Forward Flows` as weekly, while older `/dashboard` JSON responses reported it as daily. The feed configuration therefore binds that source to `WEEKLY`; stale alerts display the effective cadence used by the monitor.
+
+The four known 3Jane sources are required, and a missing or malformed freshness record for one of them makes the feed **stale**, not unavailable. Freshness can no longer be established, but the collateral ratio itself is unaffected — so the report is still returned and the sub-95% check still runs. An upstream source rename degrades the feed to a MEDIUM staleness alert; it cannot silently disable the CRITICAL solvency check.
+
+### Ratio alerts
+
+HIGH fires once when the ratio drops below 99%; CRITICAL fires once when it stays below 95% for **two consecutive, newer reports**. Each severity stays quiet until the ratio recovers above its threshold. Re-polling a frozen report cannot confirm CRITICAL, and an unavailable run resets partial confirmation. A single sub-95% reading is reported as HIGH so it stays visible without escalating on what is more likely a stale document-report refresh.
+
+The 95%/99% bands are temporary test thresholds while Accountable's report excludes 3Jane idle funds. Recalibrate both thresholds when idle funds are included in the reported reserve totals.
+
+### No emergency dispatch
+
+Accountable alerts are sent with protocol key `3jane-accountable` and channel `3jane`. They reach the normal 3Jane Telegram channel at full severity, but the protocol key is deliberately **absent** from `utils.dispatch.DISPATCHABLE_PROTOCOLS`, so a CRITICAL here cannot trigger the emergency cap-zeroing webhook.
+
+This is intentional for v1: the live collateral margin is only a few basis points, and the feed's noise profile needs a burn-in period before it should be allowed to drive automated action. Revisit once there is enough operating history — see issue #327.
+
 ## Alert dispatch
 
 Alerts use the structured `send_alert` path. HIGH and CRITICAL alerts invoke the default emergency-dispatch hook after Telegram delivery, and `3jane` is enabled in `utils.dispatch.DISPATCHABLE_PROTOCOLS`.
diff --git a/protocols/3jane/main.py b/protocols/3jane/main.py
index 691ce064..f7b8eb0a 100644
--- a/protocols/3jane/main.py
+++ b/protocols/3jane/main.py
@@ -20,6 +20,9 @@
 - Debt cap changes — alerts when ProtocolConfig debt cap is modified
 - Nominal sUSD3 backing floor — alerts on change and when floor > sUSD3 backing
 - Protocol-wide pause — alerts once when ProtocolConfig IS_PAUSED flips to true
+- Accountable Proof of Solvency — collateral ratio thresholds plus feed freshness
+  and availability. Alerts route to the 3Jane channel but never trigger the
+  emergency dispatch webhook; see the README for why.
 """
 
 import json
@@ -28,11 +31,13 @@
 import urllib.request
 from dataclasses import dataclass
 from datetime import datetime, timezone
+from decimal import Decimal
 from typing import Any
 
 from web3 import Web3
 
 from utils.abi import load_abi
+from utils.accountable import AccountableFeedConfig, AccountableReport, AccountableStatus, fetch_report
 from utils.alert import Alert, AlertSeverity, send_alert
 from utils.cache import cache_path, get_last_value_for_key_from_file, write_last_value_to_file
 from utils.chains import Chain
@@ -44,6 +49,13 @@
 PROTOCOL = "3jane"
 logger = get_logger(PROTOCOL)
 
+# Accountable alerts route to the 3Jane Telegram channel but deliberately use a
+# protocol key that is absent from utils.dispatch.DISPATCHABLE_PROTOCOLS, so a
+# CRITICAL here cannot trigger the emergency cap-zeroing webhook. The live
+# collateral margin is only a few basis points, so automated action on this
+# signal needs a burn-in period first. See issue #327.
+ACCOUNTABLE_ALERT_PROTOCOL = "3jane-accountable"
+
 CACHE_FILENAME = cache_path("cache-id.txt")
 
 # --- ABIs ---
@@ -84,6 +96,13 @@
 CACHE_KEY_JUNIOR_BUFFER_ALERTED = "3JANE_JUNIOR_BUFFER_ALERTED"
 CACHE_KEY_USD3_OC_ALERTED = "3JANE_USD3_OC_ALERTED"
 CACHE_KEY_WITHDRAW_LIMIT_ALERTED = "3JANE_WITHDRAW_LIMIT_ALERTED"
+CACHE_KEY_ACCOUNTABLE_HIGH_ALERTED = "3JANE_ACCOUNTABLE_HIGH_ALERTED"
+CACHE_KEY_ACCOUNTABLE_CRITICAL_ALERTED = "3JANE_ACCOUNTABLE_CRITICAL_ALERTED"
+CACHE_KEY_ACCOUNTABLE_CRITICAL_STREAK = "3JANE_ACCOUNTABLE_CRITICAL_STREAK"
+CACHE_KEY_ACCOUNTABLE_CRITICAL_LAST_TS = "3JANE_ACCOUNTABLE_CRITICAL_LAST_TS"
+CACHE_KEY_ACCOUNTABLE_FAILURE_STREAK = "3JANE_ACCOUNTABLE_FAILURE_STREAK"
+CACHE_KEY_ACCOUNTABLE_HEALTH_ALERTED = "3JANE_ACCOUNTABLE_HEALTH_ALERTED"
+CACHE_KEY_ACCOUNTABLE_STALE_ALERTED = "3JANE_ACCOUNTABLE_STALE_ALERTED"
 
 # --- ProtocolConfig keys (keccak256 of the string label) ---
 CFG_KEY_SUSD3_NOMINAL_BACKING_FLOOR = Web3.keccak(text="SUSD3_NOMINAL_BACKING_FLOOR")
@@ -97,6 +116,28 @@
 INSURANCE_FUND_OUTFLOW_THRESHOLD = 50_000  # USDC
 WITHDRAW_LIMIT_THRESHOLD = 4_000_000  # USDC, alert when USD3 availableWithdrawLimit falls below
 
+# --- Accountable Proof of Solvency ---
+ACCOUNTABLE_FEED = AccountableFeedConfig(
+    dfid="100000026",
+    dashboard_url=os.getenv("THREE_JANE_ACCOUNTABLE_URL", "https://accountable.3jane.xyz/dashboard"),
+    message_url=os.getenv("THREE_JANE_ACCOUNTABLE_MESSAGE_URL", "https://accountable.3jane.xyz/"),
+    dashboard_type="three-jane",
+    required_sources=(
+        "LendSwift - Warehouse Senior Note",
+        "USD3 Minted Liabilities",
+        "Slope - Forward Flows",
+        "USD3 On-Chain Reserves",
+    ),
+    # The dashboard UI declares Slope as Weekly, while older JSON responses
+    # reported Daily. Bind the operator-confirmed cadence to avoid false alerts.
+    source_frequency_overrides=(("Slope - Forward Flows", "WEEKLY"),),
+)
+# TODO: Recalibrate both thresholds after Accountable includes 3Jane's idle
+# funds in the reported reserve totals. These values are temporary test bands.
+ACCOUNTABLE_CRITICAL_RATIO = Decimal("0.95")
+ACCOUNTABLE_HIGH_RATIO = Decimal("0.99")
+ACCOUNTABLE_CRITICAL_CONFIRMATIONS = 2
+
 THREE_JANE_BORROWER_DEFAULT_WATCH_QUERY = """
 query GetThreeJaneBorrowerDefaultWatch($limit: Int!, $offset: Int!) {
   ThreeJaneBorrowerMarket(
@@ -163,8 +204,8 @@ def get_cache_int(key: str) -> int:
             return 0
 
 
-def set_cache_value(key: str, value: int | float) -> None:
-    """Write a numeric value to cache."""
+def set_cache_value(key: str, value: int | float | str) -> None:
+    """Write a value to cache (numeric, or a label such as an alert band)."""
     write_last_value_to_file(CACHE_FILENAME, key, value)
 
 
@@ -900,10 +941,230 @@ def check_protocol_paused(is_paused: bool) -> None:
         set_cache_value(CACHE_KEY_IS_PAUSED, float(is_paused))
 
 
+def _accountable_alert(severity: AlertSeverity, message: str) -> None:
+    """Send an Accountable alert on the 3Jane channel without emergency dispatch."""
+    send_alert(Alert(severity, message, ACCOUNTABLE_ALERT_PROTOCOL, channel=PROTOCOL))
+
+
+def _reset_accountable_critical_confirmation() -> None:
+    """Clear partial CRITICAL confirmation after a gap or non-critical report."""
+    if get_cache_int(CACHE_KEY_ACCOUNTABLE_CRITICAL_STREAK):
+        set_cache_value(CACHE_KEY_ACCOUNTABLE_CRITICAL_STREAK, 0)
+    if get_cache_int(CACHE_KEY_ACCOUNTABLE_CRITICAL_LAST_TS):
+        set_cache_value(CACHE_KEY_ACCOUNTABLE_CRITICAL_LAST_TS, 0)
+
+
+def _clear_accountable_ratio_alerts() -> None:
+    """Re-arm HIGH/CRITICAL ratio alerts after recovery above the warning threshold."""
+    if get_cache_int(CACHE_KEY_ACCOUNTABLE_HIGH_ALERTED):
+        set_cache_value(CACHE_KEY_ACCOUNTABLE_HIGH_ALERTED, 0)
+    if get_cache_int(CACHE_KEY_ACCOUNTABLE_CRITICAL_ALERTED):
+        set_cache_value(CACHE_KEY_ACCOUNTABLE_CRITICAL_ALERTED, 0)
+
+
+def _critical_confirmed(report_ts_ms: int) -> bool:
+    """Return True once enough consecutive newer sub-critical reports are seen.
+
+    A first reading is treated as HIGH so it stays visible. Re-polling a frozen
+    report cannot confirm itself.
+    """
+    streak = get_cache_int(CACHE_KEY_ACCOUNTABLE_CRITICAL_STREAK)
+    last_ts_ms = get_cache_int(CACHE_KEY_ACCOUNTABLE_CRITICAL_LAST_TS)
+    if report_ts_ms > last_ts_ms:
+        streak += 1
+        set_cache_value(CACHE_KEY_ACCOUNTABLE_CRITICAL_STREAK, streak)
+        set_cache_value(CACHE_KEY_ACCOUNTABLE_CRITICAL_LAST_TS, report_ts_ms)
+    else:
+        logger.info(
+            "Accountable collateral remains below %s on unchanged/out-of-order report %d (last %d); "
+            "confirmation stays at %d/%d",
+            ACCOUNTABLE_CRITICAL_RATIO,
+            report_ts_ms,
+            last_ts_ms,
+            streak,
+            ACCOUNTABLE_CRITICAL_CONFIRMATIONS,
+        )
+    if streak >= ACCOUNTABLE_CRITICAL_CONFIRMATIONS:
+        return True
+
+    logger.info(
+        "Accountable collateral below %s but unconfirmed (%d/%d runs); holding at HIGH",
+        ACCOUNTABLE_CRITICAL_RATIO,
+        streak,
+        ACCOUNTABLE_CRITICAL_CONFIRMATIONS,
+    )
+    return False
+
+
+def _format_accountable_report(report: AccountableReport) -> str:
+    """Render the shared report body used by every Accountable alert."""
+    return (
+        f"📊 Collateral ratio: {report.collateralization:.4%}\n"
+        f"💰 Reserves: {format_usd(float(report.total_reserves))} | "
+        f"Liabilities: {format_usd(float(report.total_supply))}\n"
+        f"🧮 Net: {format_usd(float(report.net))} | Verifiability: {report.verifiability}%\n"
+        f"🕒 Report: {report.report_timestamp:%Y-%m-%d %H:%M:%S UTC} "
+        f"({format_duration(report.report_age_seconds)} old)"
+    )
+
+
+def _alert_accountable_high(report: AccountableReport) -> None:
+    """Alert once while ratio is below the HIGH threshold."""
+    if get_cache_int(CACHE_KEY_ACCOUNTABLE_HIGH_ALERTED):
+        return
+    message = (
+        f"🚨 *3Jane Proof of Solvency Low*\n"
+        f"{_format_accountable_report(report)}\n"
+        f"⚠️ Collateral ratio below the {ACCOUNTABLE_HIGH_RATIO:.0%} warning threshold\n"
+        f"🔗 [Accountable dashboard]({ACCOUNTABLE_FEED.message_url})"
+    )
+    _accountable_alert(AlertSeverity.HIGH, message)
+    set_cache_value(CACHE_KEY_ACCOUNTABLE_HIGH_ALERTED, 1)
+
+
+def _alert_accountable_critical(report: AccountableReport) -> None:
+    """Alert once while ratio is confirmed below the critical threshold."""
+    if get_cache_int(CACHE_KEY_ACCOUNTABLE_CRITICAL_ALERTED):
+        return
+    message = (
+        f"🚨 *3Jane Proof of Solvency CRITICAL*\n"
+        f"{_format_accountable_report(report)}\n"
+        f"⚠️ Collateral ratio below the {ACCOUNTABLE_CRITICAL_RATIO:.0%} critical threshold\n"
+        f"🔗 [Accountable dashboard]({ACCOUNTABLE_FEED.message_url})"
+    )
+    _accountable_alert(AlertSeverity.CRITICAL, message)
+    set_cache_value(CACHE_KEY_ACCOUNTABLE_CRITICAL_ALERTED, 1)
+    # Avoid a follow-up HIGH once CRITICAL clears but ratio is still under the HIGH threshold.
+    set_cache_value(CACHE_KEY_ACCOUNTABLE_HIGH_ALERTED, 1)
+
+
+def check_accountable_collateral(report: AccountableReport) -> None:
+    """Alert when Accountable collateral ratio breaches thresholds.
+
+    HIGH when ratio < 99%; CRITICAL when ratio < 95% for two consecutive newer
+    reports. Each severity alerts once until the ratio recovers above its threshold.
+
+    Args:
+        report: Validated Proof of Solvency report.
+    """
+    ratio = report.collateralization
+    logger.info("Accountable collateral ratio: %.6f%%", ratio * 100)
+
+    if ratio < ACCOUNTABLE_CRITICAL_RATIO:
+        if _critical_confirmed(report.ts_ms):
+            _alert_accountable_critical(report)
+        else:
+            _alert_accountable_high(report)
+        return
+
+    if ratio < ACCOUNTABLE_HIGH_RATIO:
+        _reset_accountable_critical_confirmation()
+        if get_cache_int(CACHE_KEY_ACCOUNTABLE_CRITICAL_ALERTED):
+            set_cache_value(CACHE_KEY_ACCOUNTABLE_CRITICAL_ALERTED, 0)
+        _alert_accountable_high(report)
+        return
+
+    _reset_accountable_critical_confirmation()
+    _clear_accountable_ratio_alerts()
+
+
+def check_accountable_staleness(report: AccountableReport, reason: str) -> None:
+    """Alert once when the Accountable feed or one of its sources goes stale.
+
+    A fresh aggregate timestamp does not prove every input is fresh, and the
+    reserves side leans heavily on manually uploaded document reports. Deduped
+    until the feed recovers.
+
+    Args:
+        report: Report the staleness was detected on.
+        reason: Human-readable description of what is stale.
+    """
+    logger.warning("Accountable feed %s is stale: %s", ACCOUNTABLE_FEED.dfid, reason)
+
+    if get_cache_int(CACHE_KEY_ACCOUNTABLE_STALE_ALERTED):
+        return
+
+    message = (
+        f"⚠️ *3Jane Proof of Solvency Stale*\n"
+        f"{_format_accountable_report(report)}\n"
+        f"🕳️ {escape_markdown(reason)}\n"
+        f"⚠️ Collateral ratio may not reflect current positions\n"
+        f"🔗 [Accountable dashboard]({ACCOUNTABLE_FEED.message_url})"
+    )
+    _accountable_alert(AlertSeverity.MEDIUM, message)
+    set_cache_value(CACHE_KEY_ACCOUNTABLE_STALE_ALERTED, 1)
+
+
+def check_accountable_availability(reason: str) -> None:
+    """Track feed failures and alert after one exhausted retrieval cycle.
+
+    ``fetch_report`` has already exhausted bounded retries before reporting a
+    request failure. The alert is deduplicated until the feed recovers.
+
+    Args:
+        reason: Why the feed was unusable this run.
+    """
+    # An unusable run breaks the sequence of confirmed collateral observations.
+    _reset_accountable_critical_confirmation()
+
+    streak = get_cache_int(CACHE_KEY_ACCOUNTABLE_FAILURE_STREAK) + 1
+    set_cache_value(CACHE_KEY_ACCOUNTABLE_FAILURE_STREAK, streak)
+    logger.warning("Accountable feed unusable (%d consecutive): %s", streak, reason)
+
+    if get_cache_int(CACHE_KEY_ACCOUNTABLE_HEALTH_ALERTED):
+        return
+
+    message = (
+        f"⚠️ *3Jane Proof of Solvency Unavailable*\n"
+        f"📡 Retrieval failed after all retry attempts\n"
+        f"❌ {escape_markdown(reason)}\n"
+        f"⚠️ Collateral ratio is not being monitored\n"
+        f"🔗 [Accountable dashboard]({ACCOUNTABLE_FEED.message_url})"
+    )
+    _accountable_alert(AlertSeverity.HIGH, message)
+    set_cache_value(CACHE_KEY_ACCOUNTABLE_HEALTH_ALERTED, 1)
+
+
+def check_accountable_solvency() -> None:
+    """Fetch and evaluate the Accountable Proof of Solvency feed.
+
+    Runs in its own failure boundary: any error here is logged and swallowed so
+    the onchain 3Jane checks always complete.
+    """
+    try:
+        result = fetch_report(ACCOUNTABLE_FEED)
+
+        if result.status is AccountableStatus.UNAVAILABLE or result.report is None:
+            check_accountable_availability(result.reason or "unknown error")
+            return
+
+        # Reachable and parseable: clear any outstanding availability alert.
+        if get_cache_int(CACHE_KEY_ACCOUNTABLE_FAILURE_STREAK):
+            set_cache_value(CACHE_KEY_ACCOUNTABLE_FAILURE_STREAK, 0)
+        if get_cache_int(CACHE_KEY_ACCOUNTABLE_HEALTH_ALERTED):
+            set_cache_value(CACHE_KEY_ACCOUNTABLE_HEALTH_ALERTED, 0)
+
+        report = result.report
+        if result.status is AccountableStatus.STALE:
+            check_accountable_staleness(report, result.reason)
+        elif get_cache_int(CACHE_KEY_ACCOUNTABLE_STALE_ALERTED):
+            set_cache_value(CACHE_KEY_ACCOUNTABLE_STALE_ALERTED, 0)
+
+        # The ratio is still evaluated on a stale report: a low-ratio
+        # reading matters even when the inputs behind it have aged.
+        check_accountable_collateral(report)
+    except Exception as e:
+        logger.error("Error during Accountable Proof of Solvency check: %s", e)
+
+
 def main() -> None:
     """Run all 3Jane monitoring checks."""
     logger.info("Starting 3Jane monitoring...")
 
+    # Runs before the onchain reads and inside its own failure boundary, so the
+    # solvency feed is checked even when RPC access is degraded.
+    check_accountable_solvency()
+
     client = ChainManager.get_client(Chain.MAINNET)
     usd3_vault = client.eth.contract(address=USD3_ADDRESS, abi=ABI_VAULT)
     susd3_vault = client.eth.contract(address=SUSD3_ADDRESS, abi=ABI_VAULT)
diff --git a/tests/fixtures/accountable_3jane_dashboard.json b/tests/fixtures/accountable_3jane_dashboard.json
new file mode 100644
index 00000000..c2fd93ba
--- /dev/null
+++ b/tests/fixtures/accountable_3jane_dashboard.json
@@ -0,0 +1,42 @@
+{
+  "res": "ok",
+  "data": {
+    "collateralization": 1.000288,
+    "net": 21565.08,
+    "ts": "1785490814726",
+    "reserves": {
+      "verifiability": "100",
+      "interval": "live",
+      "total_reserves": {
+        "value": 75021555.53,
+        "name": "Total Reserves"
+      },
+      "total_supply": {
+        "value": 74999990.44,
+        "name": "Total Supply"
+      }
+    },
+    "dataSources": {
+      "LendSwift - Warehouse Senior Note": {
+        "type": "Document Report",
+        "frequency": "WEEKLY",
+        "lastUpdated": "1784851200000"
+      },
+      "USD3 Minted Liabilities": {
+        "type": "ERC4626",
+        "frequency": "15 MIN",
+        "lastUpdated": "1785490200523"
+      },
+      "Slope - Forward Flows": {
+        "type": "Document Report",
+        "frequency": "DAILY",
+        "lastUpdated": "1785283200000"
+      },
+      "USD3 On-Chain Reserves": {
+        "type": "ERC4626",
+        "frequency": "15 MIN",
+        "lastUpdated": "1785490201173"
+      }
+    }
+  }
+}
diff --git a/tests/test_3jane.py b/tests/test_3jane.py
index facd4987..f4e4677e 100644
--- a/tests/test_3jane.py
+++ b/tests/test_3jane.py
@@ -501,3 +501,278 @@ def test_parse_envio_borrower_default_watch_rows_default_started_forces_default(
     assert parsed[0].repayment_status == "Default"
     assert parsed[0].default_bucket == "default"
     assert parsed[0].seconds_since_default == 0
+
+
+# --- Accountable Proof of Solvency ---
+
+
+def make_accountable_report(
+    module: ModuleType,
+    ratio: str,
+    age_seconds: int = 300,
+    ts_ms: int = 1_785_490_814_726,
+):
+    """Build a report whose totals produce the requested collateral ratio."""
+    from decimal import Decimal
+
+    from utils.accountable import AccountableReport
+
+    supply = Decimal("75000000")
+    reserves = supply * Decimal(ratio)
+    return AccountableReport(
+        dfid="100000026",
+        collateralization=reserves / supply,
+        reported_collateralization=round(reserves / supply, 6),
+        net=reserves - supply,
+        total_reserves=reserves,
+        total_supply=supply,
+        verifiability=Decimal("100"),
+        ts_ms=ts_ms,
+        report_age_seconds=age_seconds,
+        report_interval="live",
+        report_cadence_seconds=15 * 60,
+        sources=(),
+    )
+
+
+def test_accountable_healthy_ratio_does_not_alert(monkeypatch: pytest.MonkeyPatch) -> None:
+    module = load_3jane_module()
+    alerts: list = []
+    stub_cache(monkeypatch, module)
+    monkeypatch.setattr(module, "send_alert", alerts.append)
+
+    module.check_accountable_collateral(make_accountable_report(module, "1.20"))
+
+    assert alerts == []
+
+
+def test_accountable_high_threshold_alerts(monkeypatch: pytest.MonkeyPatch) -> None:
+    module = load_3jane_module()
+    alerts: list = []
+    stub_cache(monkeypatch, module)
+    monkeypatch.setattr(module, "send_alert", alerts.append)
+
+    module.check_accountable_collateral(make_accountable_report(module, "0.98"))
+
+    assert len(alerts) == 1
+    assert alerts[0].severity == module.AlertSeverity.HIGH
+    assert "98.0000%" in alerts[0].message
+
+
+def test_accountable_critical_requires_two_consecutive_runs(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A single sub-95% reading is more likely a stale refresh than insolvency."""
+    module = load_3jane_module()
+    alerts: list = []
+    stub_cache(monkeypatch, module)
+    monkeypatch.setattr(module, "send_alert", alerts.append)
+    first_report = make_accountable_report(module, "0.94")
+    second_report = make_accountable_report(module, "0.94", ts_ms=first_report.ts_ms + 1)
+
+    module.check_accountable_collateral(first_report)
+    assert len(alerts) == 1
+    assert alerts[0].severity == module.AlertSeverity.HIGH
+
+    module.check_accountable_collateral(second_report)
+    assert len(alerts) == 2
+    assert alerts[1].severity == module.AlertSeverity.CRITICAL
+    assert "95% critical threshold" in alerts[1].message
+
+
+def test_accountable_frozen_report_does_not_confirm_critical(monkeypatch: pytest.MonkeyPatch) -> None:
+    module = load_3jane_module()
+    alerts: list = []
+    cache = stub_cache(monkeypatch, module)
+    monkeypatch.setattr(module, "send_alert", alerts.append)
+    report = make_accountable_report(module, "0.94")
+
+    module.check_accountable_collateral(report)
+    module.check_accountable_collateral(report)
+
+    assert [alert.severity for alert in alerts] == [module.AlertSeverity.HIGH]
+    assert cache[module.CACHE_KEY_ACCOUNTABLE_CRITICAL_STREAK] == "1"
+
+
+def test_accountable_critical_streak_resets_on_recovery(monkeypatch: pytest.MonkeyPatch) -> None:
+    module = load_3jane_module()
+    alerts: list = []
+    stub_cache(monkeypatch, module)
+    monkeypatch.setattr(module, "send_alert", alerts.append)
+
+    module.check_accountable_collateral(make_accountable_report(module, "0.94"))
+    module.check_accountable_collateral(make_accountable_report(module, "1.20"))
+    module.check_accountable_collateral(make_accountable_report(module, "0.94"))
+
+    # The recovery reset the streak, so the second breach is unconfirmed again
+    # and never escalates — two non-consecutive dips must not reach CRITICAL.
+    assert [alert.severity for alert in alerts] == [module.AlertSeverity.HIGH, module.AlertSeverity.HIGH]
+
+
+def test_accountable_does_not_realert_while_below_high(monkeypatch: pytest.MonkeyPatch) -> None:
+    module = load_3jane_module()
+    alerts: list = []
+    stub_cache(monkeypatch, module)
+    monkeypatch.setattr(module, "send_alert", alerts.append)
+
+    module.check_accountable_collateral(make_accountable_report(module, "0.98"))
+    module.check_accountable_collateral(make_accountable_report(module, "0.981"))
+    module.check_accountable_collateral(make_accountable_report(module, "0.982"))
+
+    assert len(alerts) == 1
+
+
+def test_accountable_recovery_rearms_alert(monkeypatch: pytest.MonkeyPatch) -> None:
+    module = load_3jane_module()
+    alerts: list = []
+    stub_cache(monkeypatch, module)
+    monkeypatch.setattr(module, "send_alert", alerts.append)
+
+    module.check_accountable_collateral(make_accountable_report(module, "0.98"))
+    module.check_accountable_collateral(make_accountable_report(module, "1.20"))
+    module.check_accountable_collateral(make_accountable_report(module, "0.98"))
+
+    assert len(alerts) == 2
+    assert all(alert.severity == module.AlertSeverity.HIGH for alert in alerts)
+
+
+def test_accountable_alerts_never_trigger_emergency_dispatch(monkeypatch: pytest.MonkeyPatch) -> None:
+    """CRITICAL here must not zero market caps while the margin is basis points."""
+    from utils.dispatch import DISPATCHABLE_PROTOCOLS
+
+    module = load_3jane_module()
+    alerts: list = []
+    stub_cache(monkeypatch, module)
+    monkeypatch.setattr(module, "send_alert", alerts.append)
+    first_report = make_accountable_report(module, "0.94")
+    second_report = make_accountable_report(module, "0.94", ts_ms=first_report.ts_ms + 1)
+
+    module.check_accountable_collateral(first_report)
+    module.check_accountable_collateral(second_report)
+
+    assert alerts[-1].severity == module.AlertSeverity.CRITICAL
+    assert alerts[-1].protocol == module.ACCOUNTABLE_ALERT_PROTOCOL
+    assert alerts[-1].protocol not in DISPATCHABLE_PROTOCOLS
+    # Still routed to the normal 3Jane Telegram channel.
+    assert alerts[-1].channel == module.PROTOCOL
+    assert module.PROTOCOL in DISPATCHABLE_PROTOCOLS
+
+
+def test_accountable_unavailable_run_breaks_critical_confirmation(monkeypatch: pytest.MonkeyPatch) -> None:
+    module = load_3jane_module()
+    alerts: list = []
+    cache = stub_cache(monkeypatch, module)
+    monkeypatch.setattr(module, "send_alert", alerts.append)
+    first_ts = 1_785_490_814_726
+
+    module.check_accountable_collateral(make_accountable_report(module, "0.94", ts_ms=first_ts))
+    module.check_accountable_availability("connection refused")
+    module.check_accountable_collateral(make_accountable_report(module, "0.94", ts_ms=first_ts + 1))
+
+    assert [alert.severity for alert in alerts] == [
+        module.AlertSeverity.HIGH,  # Initial ratio warning.
+        module.AlertSeverity.HIGH,  # Feed unavailable after retries.
+    ]
+    assert cache[module.CACHE_KEY_ACCOUNTABLE_CRITICAL_STREAK] == "1"
+
+    module.check_accountable_collateral(make_accountable_report(module, "0.94", ts_ms=first_ts + 2))
+    assert [alert.severity for alert in alerts] == [
+        module.AlertSeverity.HIGH,
+        module.AlertSeverity.HIGH,
+        module.AlertSeverity.CRITICAL,
+    ]
+
+
+def test_accountable_staleness_alerts_once(monkeypatch: pytest.MonkeyPatch) -> None:
+    module = load_3jane_module()
+    alerts: list = []
+    stub_cache(monkeypatch, module)
+    monkeypatch.setattr(module, "send_alert", alerts.append)
+    report = make_accountable_report(module, "1.20")
+
+    module.check_accountable_staleness(report, "stale sources: Slope (99h)")
+    module.check_accountable_staleness(report, "stale sources: Slope (100h)")
+
+    assert len(alerts) == 1
+    assert alerts[0].severity == module.AlertSeverity.MEDIUM
+    assert "Slope" in alerts[0].message
+
+
+def test_accountable_availability_alerts_after_retries_are_exhausted(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    module = load_3jane_module()
+    alerts: list = []
+    stub_cache(monkeypatch, module)
+    monkeypatch.setattr(module, "send_alert", alerts.append)
+
+    module.check_accountable_availability("connection refused")
+    assert len(alerts) == 1
+    assert alerts[0].severity == module.AlertSeverity.HIGH
+    assert "after all retry attempts" in alerts[0].message
+
+    # Stays quiet while the outage persists.
+    module.check_accountable_availability("connection refused")
+    assert len(alerts) == 1
+
+
+def test_accountable_stale_report_still_evaluates_ratio(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A low collateralization reading matters even when its inputs have aged."""
+    from utils.accountable import AccountableFetchResult, AccountableStatus
+
+    module = load_3jane_module()
+    alerts: list = []
+    stub_cache(monkeypatch, module)
+    monkeypatch.setattr(module, "send_alert", alerts.append)
+    report = make_accountable_report(module, "0.98")
+    monkeypatch.setattr(
+        module,
+        "fetch_report",
+        lambda _config: AccountableFetchResult(AccountableStatus.STALE, report, "stale sources: Slope (99h)"),
+    )
+
+    module.check_accountable_solvency()
+
+    severities = [alert.severity for alert in alerts]
+    assert module.AlertSeverity.MEDIUM in severities  # staleness
+    assert module.AlertSeverity.HIGH in severities  # low-ratio reading
+
+
+def test_accountable_recovery_clears_health_alert(monkeypatch: pytest.MonkeyPatch) -> None:
+    from utils.accountable import AccountableFetchResult, AccountableStatus
+
+    module = load_3jane_module()
+    alerts: list = []
+    cache = stub_cache(monkeypatch, module)
+    monkeypatch.setattr(module, "send_alert", alerts.append)
+    monkeypatch.setattr(
+        module,
+        "fetch_report",
+        lambda _config: AccountableFetchResult(AccountableStatus.UNAVAILABLE, None, "boom"),
+    )
+
+    module.check_accountable_solvency()
+    assert len(alerts) == 1
+    assert cache[module.CACHE_KEY_ACCOUNTABLE_HEALTH_ALERTED] == "1"
+
+    monkeypatch.setattr(
+        module,
+        "fetch_report",
+        lambda _config: AccountableFetchResult(AccountableStatus.OK, make_accountable_report(module, "1.20")),
+    )
+    module.check_accountable_solvency()
+
+    assert cache[module.CACHE_KEY_ACCOUNTABLE_HEALTH_ALERTED] == "0"
+    assert cache[module.CACHE_KEY_ACCOUNTABLE_FAILURE_STREAK] == "0"
+
+
+def test_accountable_failure_does_not_raise(monkeypatch: pytest.MonkeyPatch) -> None:
+    """The onchain checks must survive any Accountable-side explosion."""
+    module = load_3jane_module()
+    stub_cache(monkeypatch, module)
+    monkeypatch.setattr(module, "send_alert", lambda _alert: None)
+
+    def boom(_config):
+        raise RuntimeError("unexpected")
+
+    monkeypatch.setattr(module, "fetch_report", boom)
+
+    module.check_accountable_solvency()  # must not raise
diff --git a/tests/test_accountable.py b/tests/test_accountable.py
new file mode 100644
index 00000000..5d3216c8
--- /dev/null
+++ b/tests/test_accountable.py
@@ -0,0 +1,475 @@
+"""Tests for the Accountable Proof of Solvency client."""
+
+import copy
+import json
+from dataclasses import replace
+from decimal import Decimal
+from pathlib import Path
+from typing import Any
+
+import pytest
+import requests
+
+from utils import accountable
+from utils.accountable import (
+    AccountableError,
+    AccountableFeedConfig,
+    AccountableStatus,
+    evaluate_report,
+    fetch_report,
+    parse_frequency_seconds,
+    parse_report,
+)
+
+FIXTURE_PATH = Path(__file__).parent / "fixtures" / "accountable_3jane_dashboard.json"
+# Fixture was recorded at ts=1785490814726; this is a few minutes later.
+FIXTURE_NOW_MS = 1785491000000
+
+CONFIG = AccountableFeedConfig(
+    dfid="100000026",
+    dashboard_url="https://accountable.3jane.xyz/dashboard",
+    message_url="https://accountable.3jane.xyz/",
+    dashboard_type="three-jane",
+    required_sources=(
+        "LendSwift - Warehouse Senior Note",
+        "USD3 Minted Liabilities",
+        "Slope - Forward Flows",
+        "USD3 On-Chain Reserves",
+    ),
+    source_frequency_overrides=(("Slope - Forward Flows", "WEEKLY"),),
+)
+
+
+def load_payload() -> dict[str, Any]:
+    """Return a mutable copy of the recorded dashboard response."""
+    return json.loads(FIXTURE_PATH.read_text())
+
+
+def load_fresh_payload() -> dict[str, Any]:
+    """Return the recorded report with every required source made current."""
+    payload = load_payload()
+    for source in payload["data"]["dataSources"].values():
+        source["lastUpdated"] = str(FIXTURE_NOW_MS)
+    return payload
+
+
+# --- Parsing the real recorded payload ---
+
+
+def test_parses_recorded_live_payload() -> None:
+    report = parse_report(load_payload(), CONFIG, FIXTURE_NOW_MS)
+
+    assert report.dfid == "100000026"
+    assert report.total_reserves == Decimal("75021555.53")
+    assert report.total_supply == Decimal("74999990.44")
+    assert report.net == Decimal("21565.08")
+    assert report.verifiability == Decimal("100")
+    assert report.ts_ms == 1785490814726
+
+
+def test_ratio_is_derived_at_full_precision_not_taken_from_rounded_field() -> None:
+    report = parse_report(load_payload(), CONFIG, FIXTURE_NOW_MS)
+
+    assert report.reported_collateralization == Decimal("1.000288")
+    assert report.collateralization == Decimal("75021555.53") / Decimal("74999990.44")
+    # The derived value carries precision the rounded API field discards.
+    assert report.collateralization != report.reported_collateralization
+
+
+def test_coerces_numeric_strings() -> None:
+    """ts and verifiability ship as strings despite being documented as numbers."""
+    payload = load_payload()
+    assert isinstance(payload["data"]["ts"], str)
+    assert isinstance(payload["data"]["reserves"]["verifiability"], str)
+
+    report = parse_report(payload, CONFIG, FIXTURE_NOW_MS)
+
+    assert isinstance(report.ts_ms, int)
+    assert report.verifiability == Decimal("100")
+
+
+def test_recorded_live_payload_uses_slope_weekly_override() -> None:
+    result = evaluate_report(parse_report(load_payload(), CONFIG, FIXTURE_NOW_MS))
+
+    assert result.status is AccountableStatus.STALE
+    assert result.report is not None
+    assert [source.name for source in result.report.stale_sources] == [
+        "LendSwift - Warehouse Senior Note",
+    ]
+    slope = next(source for source in result.report.sources if source.name == "Slope - Forward Flows")
+    assert slope.frequency == "WEEKLY"
+    assert not slope.is_stale
+
+
+# --- Rejection cases ---
+
+
+def test_rejects_non_ok_res() -> None:
+    payload = load_payload()
+    payload["res"] = "error"
+
+    with pytest.raises(AccountableError, match="res"):
+        parse_report(payload, CONFIG, FIXTURE_NOW_MS)
+
+
+def test_rejects_missing_reserves() -> None:
+    payload = load_payload()
+    del payload["data"]["reserves"]
+
+    with pytest.raises(AccountableError, match="reserves"):
+        parse_report(payload, CONFIG, FIXTURE_NOW_MS)
+
+
+def test_rejects_missing_total_supply_value() -> None:
+    payload = load_payload()
+    del payload["data"]["reserves"]["total_supply"]["value"]
+
+    with pytest.raises(AccountableError, match="total_supply.value"):
+        parse_report(payload, CONFIG, FIXTURE_NOW_MS)
+
+
+def test_rejects_non_numeric_field() -> None:
+    payload = load_payload()
+    payload["data"]["reserves"]["total_reserves"]["value"] = "not-a-number"
+
+    with pytest.raises(AccountableError, match="not numeric"):
+        parse_report(payload, CONFIG, FIXTURE_NOW_MS)
+
+
+def test_rejects_non_finite_value() -> None:
+    payload = load_payload()
+    payload["data"]["reserves"]["total_reserves"]["value"] = float("inf")
+
+    with pytest.raises(AccountableError, match="not finite"):
+        parse_report(payload, CONFIG, FIXTURE_NOW_MS)
+
+
+def test_rejects_zero_supply_instead_of_dividing_by_zero() -> None:
+    payload = load_payload()
+    payload["data"]["reserves"]["total_supply"]["value"] = 0
+
+    with pytest.raises(AccountableError, match="implausible totals"):
+        parse_report(payload, CONFIG, FIXTURE_NOW_MS)
+
+
+def test_rejects_inconsistent_collateralization() -> None:
+    """A reported ratio that disagrees with the totals means we cannot trust either."""
+    payload = load_payload()
+    payload["data"]["collateralization"] = 1.5
+
+    with pytest.raises(AccountableError, match="disagrees"):
+        parse_report(payload, CONFIG, FIXTURE_NOW_MS)
+
+
+def test_rejects_inconsistent_net() -> None:
+    payload = load_payload()
+    payload["data"]["net"] = 999_999.0
+
+    with pytest.raises(AccountableError, match="net"):
+        parse_report(payload, CONFIG, FIXTURE_NOW_MS)
+
+
+def test_accepts_rounding_level_disagreement() -> None:
+    """The API rounds collateralization to 6dp, so tolerance must absorb that."""
+    payload = load_payload()
+    reserves = Decimal(str(payload["data"]["reserves"]["total_reserves"]["value"]))
+    supply = Decimal(str(payload["data"]["reserves"]["total_supply"]["value"]))
+    payload["data"]["collateralization"] = float(round(reserves / supply, 6))
+
+    report = parse_report(payload, CONFIG, FIXTURE_NOW_MS)
+
+    assert report.collateralization == reserves / supply
+
+
+def test_rejects_non_usd_pegged_feed() -> None:
+    """liabilities == total_supply only holds when fx is 1."""
+    payload = load_payload()
+    payload["data"]["reserves"]["total_supply"]["fx"] = 0.92
+
+    with pytest.raises(AccountableError, match="fx"):
+        parse_report(payload, CONFIG, FIXTURE_NOW_MS)
+
+
+def test_rejects_non_usd_liabilities_when_fx_is_omitted() -> None:
+    """With fx absent, the net identity is what enforces the USD denominator."""
+    payload = load_payload()
+    reserves = Decimal(str(payload["data"]["reserves"]["total_reserves"]["value"]))
+    supply = Decimal(str(payload["data"]["reserves"]["total_supply"]["value"]))
+    liabilities = supply * Decimal("0.92")
+    payload["data"]["net"] = float(reserves - liabilities)
+    payload["data"]["collateralization"] = float(round(reserves / liabilities, 6))
+    assert "fx" not in payload["data"]["reserves"]["total_supply"]
+
+    # Both consistency identities are computed against raw supply, so a feed
+    # denominated in anything else fails on the ratio cross-check first.
+    with pytest.raises(AccountableError, match="disagrees"):
+        parse_report(payload, CONFIG, FIXTURE_NOW_MS)
+
+
+def test_rejects_future_timestamp() -> None:
+    payload = load_payload()
+    payload["data"]["ts"] = str(FIXTURE_NOW_MS + 86_400_000)
+
+    with pytest.raises(AccountableError, match="future"):
+        parse_report(payload, CONFIG, FIXTURE_NOW_MS)
+
+
+@pytest.mark.parametrize("interval", [None, "sometimes"])
+def test_rejects_unrecognised_report_interval(interval: str | None) -> None:
+    payload = load_payload()
+    payload["data"]["reserves"]["interval"] = interval
+
+    with pytest.raises(AccountableError, match="reserves.interval"):
+        parse_report(payload, CONFIG, FIXTURE_NOW_MS)
+
+
+# --- Freshness ---
+
+
+@pytest.mark.parametrize(
+    ("text", "expected"),
+    [
+        ("15 MIN", 900),
+        ("DAILY", 86_400),
+        ("WEEKLY", 604_800),
+        ("48 H", 172_800),
+        ("1 W", 604_800),
+        ("live", 900),
+        ("", None),
+        ("sometimes", None),
+        (None, None),
+        # Ambiguous between minutes and months; guessing months would grant a
+        # 30x freshness budget, so it is rejected instead.
+        ("15 M", None),
+    ],
+)
+def test_parse_frequency_seconds(text: Any, expected: int | None) -> None:
+    assert parse_frequency_seconds(text) == expected
+
+
+def test_aggregate_report_is_stale_only_after_more_than_two_cadence_periods() -> None:
+    payload = load_fresh_payload()
+    payload["data"]["reserves"]["interval"] = "15 MIN"
+    payload["data"]["ts"] = str(FIXTURE_NOW_MS - 30 * 60 * 1000)
+
+    result = evaluate_report(parse_report(payload, CONFIG, FIXTURE_NOW_MS))
+
+    assert result.status is AccountableStatus.OK
+
+    payload["data"]["ts"] = str(FIXTURE_NOW_MS - (30 * 60 + 1) * 1000)
+    result = evaluate_report(parse_report(payload, CONFIG, FIXTURE_NOW_MS))
+
+    assert result.status is AccountableStatus.STALE
+    assert "old" in result.reason
+
+
+def test_weekly_aggregate_report_is_not_stale_after_six_hours() -> None:
+    payload = load_fresh_payload()
+    payload["data"]["reserves"]["interval"] = "WEEKLY"
+    payload["data"]["ts"] = str(FIXTURE_NOW_MS - 6 * 60 * 60 * 1000)
+
+    result = evaluate_report(parse_report(payload, CONFIG, FIXTURE_NOW_MS))
+
+    assert result.status is AccountableStatus.OK
+
+    payload["data"]["ts"] = str(FIXTURE_NOW_MS - (7 * 24 * 60 * 60 + 1) * 1000)
+    result = evaluate_report(parse_report(payload, CONFIG, FIXTURE_NOW_MS))
+
+    assert result.status is AccountableStatus.STALE
+
+
+def test_fresh_aggregate_with_stale_source_is_detected() -> None:
+    """The headline timestamp can be fresh while an input has gone dark."""
+    payload = load_payload()
+    payload["data"]["dataSources"]["USD3 On-Chain Reserves"]["lastUpdated"] = str(FIXTURE_NOW_MS - 86_400_000)
+
+    result = evaluate_report(parse_report(payload, CONFIG, FIXTURE_NOW_MS))
+
+    assert result.status is AccountableStatus.STALE
+    assert "USD3 On-Chain Reserves" in result.reason
+    assert result.report is not None
+    assert result.report.collateralization > 1
+
+
+def test_source_is_stale_only_after_more_than_two_cadence_periods() -> None:
+    payload = load_fresh_payload()
+    source = payload["data"]["dataSources"]["USD3 Minted Liabilities"]
+    source["lastUpdated"] = str(FIXTURE_NOW_MS - 30 * 60 * 1000)
+
+    result = evaluate_report(parse_report(payload, CONFIG, FIXTURE_NOW_MS))
+
+    assert result.status is AccountableStatus.OK
+
+    source["lastUpdated"] = str(FIXTURE_NOW_MS - (30 * 60 + 1) * 1000)
+    result = evaluate_report(parse_report(payload, CONFIG, FIXTURE_NOW_MS))
+
+    assert result.status is AccountableStatus.STALE
+    assert "USD3 Minted Liabilities" in result.reason
+
+
+def test_daily_source_is_stale_after_first_missed_period() -> None:
+    payload = load_fresh_payload()
+    source = payload["data"]["dataSources"]["USD3 Minted Liabilities"]
+    source["frequency"] = "DAILY"
+    source["lastUpdated"] = str(FIXTURE_NOW_MS - 24 * 60 * 60 * 1000)
+
+    result = evaluate_report(parse_report(payload, CONFIG, FIXTURE_NOW_MS))
+
+    assert result.status is AccountableStatus.OK
+
+    source["lastUpdated"] = str(FIXTURE_NOW_MS - (24 * 60 * 60 + 1) * 1000)
+    result = evaluate_report(parse_report(payload, CONFIG, FIXTURE_NOW_MS))
+
+    assert result.status is AccountableStatus.STALE
+    assert "USD3 Minted Liabilities" in result.reason
+
+
+def test_unparseable_source_frequency_is_skipped_not_flagged_stale() -> None:
+    payload = load_fresh_payload()
+    payload["data"]["dataSources"]["Mystery Source"] = {
+        "type": "Unknown",
+        "frequency": "whenever",
+        "lastUpdated": "1",
+    }
+
+    result = evaluate_report(parse_report(payload, CONFIG, FIXTURE_NOW_MS))
+
+    assert result.status is AccountableStatus.OK
+    assert result.report is not None
+    assert all(source.name != "Mystery Source" for source in result.report.sources)
+
+
+def test_missing_required_source_is_stale_not_unavailable() -> None:
+    """A source rename must not blind the ratio check."""
+    payload = load_payload()
+    del payload["data"]["dataSources"]["Slope - Forward Flows"]
+
+    result = evaluate_report(parse_report(payload, CONFIG, FIXTURE_NOW_MS))
+
+    assert result.status is AccountableStatus.STALE
+    assert "Slope - Forward Flows" in result.reason
+    assert result.report is not None
+    assert result.report.collateralization > 1
+
+
+@pytest.mark.parametrize("field", ["frequency", "lastUpdated", "type"])
+def test_malformed_required_source_is_stale_not_unavailable(field: str) -> None:
+    payload = load_payload()
+    del payload["data"]["dataSources"]["USD3 On-Chain Reserves"][field]
+
+    result = evaluate_report(parse_report(payload, CONFIG, FIXTURE_NOW_MS))
+
+    assert result.status is AccountableStatus.STALE
+    assert "USD3 On-Chain Reserves" in result.reason
+    assert result.report is not None
+    # The unusable source is dropped rather than counted as fresh.
+    assert all(source.name != "USD3 On-Chain Reserves" for source in result.report.sources)
+
+
+def test_future_source_timestamp_is_not_treated_as_fresh() -> None:
+    """Clamping a future lastUpdated to age 0 would defeat the freshness check."""
+    payload = load_payload()
+    payload["data"]["dataSources"]["USD3 On-Chain Reserves"]["lastUpdated"] = str(FIXTURE_NOW_MS + 86_400_000)
+
+    result = evaluate_report(parse_report(payload, CONFIG, FIXTURE_NOW_MS))
+
+    assert result.status is AccountableStatus.STALE
+    assert "future" in result.reason
+    assert result.report is not None
+    assert all(source.name != "USD3 On-Chain Reserves" for source in result.report.sources)
+
+
+def test_small_future_source_skew_is_tolerated() -> None:
+    payload = load_fresh_payload()
+    payload["data"]["dataSources"]["USD3 On-Chain Reserves"]["lastUpdated"] = str(FIXTURE_NOW_MS + 60_000)
+
+    result = evaluate_report(parse_report(payload, CONFIG, FIXTURE_NOW_MS))
+
+    assert result.status is AccountableStatus.OK
+
+
+# --- fetch_report network behaviour ---
+
+
+class _FakeResponse:
+    def __init__(self, payload: Any) -> None:
+        self._payload = payload
+
+    def json(self) -> Any:
+        if isinstance(self._payload, Exception):
+            raise self._payload
+        return self._payload
+
+
+def test_fetch_report_returns_unavailable_on_network_error(monkeypatch: pytest.MonkeyPatch) -> None:
+    def boom(*_args: Any, **_kwargs: Any) -> Any:
+        raise requests.ConnectionError("refused")
+
+    monkeypatch.setattr(accountable, "request_with_retry", boom)
+
+    result = fetch_report(CONFIG, FIXTURE_NOW_MS)
+
+    assert result.status is AccountableStatus.UNAVAILABLE
+    assert result.report is None
+    assert "refused" in result.reason
+
+
+def test_fetch_report_returns_unavailable_on_invalid_json(monkeypatch: pytest.MonkeyPatch) -> None:
+    """requests raises JSONDecodeError, which is itself a RequestException."""
+    decode_error = requests.exceptions.JSONDecodeError("not json", "<html>", 0)
+    monkeypatch.setattr(
+        accountable,
+        "request_with_retry",
+        lambda *_a, **_k: _FakeResponse(decode_error),
+    )
+
+    result = fetch_report(CONFIG, FIXTURE_NOW_MS)
+
+    assert result.status is AccountableStatus.UNAVAILABLE
+    assert "invalid JSON" in result.reason
+
+
+def test_fetch_report_rejects_non_https_url(monkeypatch: pytest.MonkeyPatch) -> None:
+    """v1 authenticates the feed with TLS alone, so plaintext must not be fetched."""
+
+    def unexpected(*_args: Any, **_kwargs: Any) -> Any:
+        raise AssertionError("request must not be attempted")
+
+    monkeypatch.setattr(accountable, "request_with_retry", unexpected)
+    config = replace(CONFIG, dashboard_url="http://accountable.3jane.xyz/dashboard")
+
+    result = fetch_report(config, FIXTURE_NOW_MS)
+
+    assert result.status is AccountableStatus.UNAVAILABLE
+    assert "HTTPS" in result.reason
+
+
+def test_fetch_report_returns_unavailable_on_schema_violation(monkeypatch: pytest.MonkeyPatch) -> None:
+    payload = load_payload()
+    del payload["data"]["collateralization"]
+    monkeypatch.setattr(accountable, "request_with_retry", lambda *_a, **_k: _FakeResponse(payload))
+
+    result = fetch_report(CONFIG, FIXTURE_NOW_MS)
+
+    assert result.status is AccountableStatus.UNAVAILABLE
+    assert result.report is None
+
+
+def test_fetch_report_success(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(accountable, "request_with_retry", lambda *_a, **_k: _FakeResponse(load_fresh_payload()))
+
+    result = fetch_report(CONFIG, FIXTURE_NOW_MS)
+
+    assert result.is_ok
+    assert result.report is not None
+    assert result.report.collateralization > 1
+
+
+def test_fixture_is_not_mutated_between_tests() -> None:
+    """Guards the shared-fixture pattern used throughout this module."""
+    first = load_payload()
+    second = load_payload()
+
+    assert first == second
+    assert first is not second
+    assert copy.deepcopy(first) == second
diff --git a/tests/test_http_client.py b/tests/test_http_client.py
new file mode 100644
index 00000000..c4f313c0
--- /dev/null
+++ b/tests/test_http_client.py
@@ -0,0 +1,86 @@
+"""Tests for shared HTTP retry behavior."""
+
+from collections.abc import Iterator
+
+import pytest
+import requests
+
+from utils import http_client
+
+
+def _response(status_code: int) -> requests.Response:
+    """Build a minimal response whose ``raise_for_status`` behaves normally."""
+    response = requests.Response()
+    response.status_code = status_code
+    response.url = "https://example.com/data"
+    return response
+
+
+def test_request_with_retry_retries_rate_limit(monkeypatch: pytest.MonkeyPatch) -> None:
+    responses: Iterator[requests.Response] = iter((_response(429), _response(200)))
+    calls: list[str] = []
+
+    def request(method: str, url: str, **_kwargs: object) -> requests.Response:
+        calls.append(f"{method}:{url}")
+        return next(responses)
+
+    monkeypatch.setattr(http_client.requests, "request", request)
+    monkeypatch.setattr(http_client.time, "sleep", lambda _seconds: None)
+
+    response = http_client.request_with_retry(
+        "get",
+        "https://example.com/data",
+        retries=1,
+        backoff_factor=0,
+        timeout=1,
+    )
+
+    assert response.status_code == 200
+    assert len(calls) == 2
+
+
+def test_request_with_retry_does_not_retry_permanent_client_error(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    calls: list[str] = []
+
+    def request(method: str, url: str, **_kwargs: object) -> requests.Response:
+        calls.append(f"{method}:{url}")
+        return _response(400)
+
+    monkeypatch.setattr(http_client.requests, "request", request)
+
+    with pytest.raises(requests.HTTPError):
+        http_client.request_with_retry(
+            "get",
+            "https://example.com/data",
+            retries=3,
+            backoff_factor=0,
+            timeout=1,
+        )
+
+    assert len(calls) == 1
+
+
+def test_request_with_retry_stops_after_rate_limit_retry_budget(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    calls: list[str] = []
+
+    def request(method: str, url: str, **_kwargs: object) -> requests.Response:
+        calls.append(f"{method}:{url}")
+        return _response(429)
+
+    monkeypatch.setattr(http_client.requests, "request", request)
+    monkeypatch.setattr(http_client.time, "sleep", lambda _seconds: None)
+
+    with pytest.raises(requests.HTTPError):
+        http_client.request_with_retry(
+            "get",
+            "https://example.com/data",
+            retries=2,
+            backoff_factor=0,
+            timeout=1,
+        )
+
+    assert len(calls) == 3
diff --git a/utils/accountable.py b/utils/accountable.py
new file mode 100644
index 00000000..7342857f
--- /dev/null
+++ b/utils/accountable.py
@@ -0,0 +1,583 @@
+"""Accountable Proof of Solvency feed client.
+
+Accountable publishes per-protocol Proof of Solvency dashboards backed by TEE
+attestations. Each dashboard exposes its full report as public JSON, which this
+module fetches, validates, and converts into a typed report.
+
+The dashboard request is URL/type-based: it neither sends nor echoes the feed id
+(DFID), so feed identity is bound explicitly through :class:`AccountableFeedConfig`.
+
+Validation is deliberately strict. The reported ``collateralization`` is rounded
+to six decimals server-side, so it is only used as a cross-check; the ratio used
+for alerting is recomputed from the underlying totals at full precision.
+
+See https://docs.accountable.capital/accountable-documentation/proof-of-solvency
+"""
+
+from __future__ import annotations
+
+import math
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from decimal import Decimal, InvalidOperation
+from enum import Enum
+from typing import Any
+
+import requests
+
+from utils.http_client import request_with_retry
+from utils.logger import get_logger
+
+logger = get_logger("utils.accountable")
+
+MS_PER_SECOND = 1000
+SECONDS_PER_HOUR = 3600
+SECONDS_PER_DAY = 86_400
+
+# The reported ratio is rounded to 6 decimals, so a cross-check can never be
+# tighter than 5e-7. Doubled to leave room for float noise in the totals.
+RATIO_CONSISTENCY_TOLERANCE = Decimal("2e-6")
+# Absolute floor for the net cross-check, scaled up for large books.
+NET_CONSISTENCY_ABSOLUTE_TOLERANCE = Decimal("1")
+NET_CONSISTENCY_RELATIVE_TOLERANCE = Decimal("1e-6")
+
+# Reserves/supply below this are treated as implausible rather than merely small:
+# a zero or negative book cannot produce a meaningful solvency ratio.
+MIN_PLAUSIBLE_TOTAL = Decimal("1")
+# A ratio outside this range indicates a schema or units problem, not solvency.
+MIN_PLAUSIBLE_RATIO = Decimal("0.01")
+MAX_PLAUSIBLE_RATIO = Decimal("100")
+# Reject reports timestamped meaningfully in the future (clock skew allowance).
+MAX_FUTURE_SKEW_SECONDS = 15 * 60
+
+# Short-cadence feeds get one missed-period allowance to absorb transient
+# delays. Longer cadences alert as soon as their first expected update is late.
+SHORT_CADENCE_MAX_SECONDS = SECONDS_PER_HOUR
+SHORT_CADENCE_STALE_PERIODS = 2
+
+# Declared cadence strings observed on Accountable dashboards, in seconds.
+_NAMED_FREQUENCIES: dict[str, int] = {
+    "LIVE": 15 * 60,
+    "REALTIME": 15 * 60,
+    "HOURLY": SECONDS_PER_HOUR,
+    "DAILY": SECONDS_PER_DAY,
+    "WEEKLY": 7 * SECONDS_PER_DAY,
+    "MONTHLY": 30 * SECONDS_PER_DAY,
+}
+_FREQUENCY_UNIT_SECONDS: dict[str, int] = {
+    "MIN": 60,
+    "MINS": 60,
+    "MINUTE": 60,
+    "MINUTES": 60,
+    "H": SECONDS_PER_HOUR,
+    "HR": SECONDS_PER_HOUR,
+    "HOUR": SECONDS_PER_HOUR,
+    "HOURS": SECONDS_PER_HOUR,
+    "D": SECONDS_PER_DAY,
+    "DAY": SECONDS_PER_DAY,
+    "DAYS": SECONDS_PER_DAY,
+    "W": 7 * SECONDS_PER_DAY,
+    "WEEK": 7 * SECONDS_PER_DAY,
+    "WEEKS": 7 * SECONDS_PER_DAY,
+    # Deliberately no bare "M": it reads as either minutes or months, and
+    # guessing months would silently hand a source a 30x freshness budget.
+    "MONTH": 30 * SECONDS_PER_DAY,
+    "MONTHS": 30 * SECONDS_PER_DAY,
+}
+
+
+def _stale_after_seconds(cadence_seconds: int) -> int:
+    """Return the age at which a report or source becomes stale."""
+    if cadence_seconds <= SHORT_CADENCE_MAX_SECONDS:
+        return cadence_seconds * SHORT_CADENCE_STALE_PERIODS
+    return cadence_seconds
+
+
+class AccountableError(Exception):
+    """Raised when an Accountable report cannot be parsed or fails validation."""
+
+
+class AccountableStatus(Enum):
+    """Outcome of an Accountable feed retrieval.
+
+    ``UNAVAILABLE`` covers every "do not trust this number" case — network
+    failure, non-200, schema violation, or a failed consistency check. ``STALE``
+    means the report parsed and is self-consistent, but the aggregate report or
+    a required source is older than its cadence allows.
+    """
+
+    OK = "OK"
+    UNAVAILABLE = "UNAVAILABLE"
+    STALE = "STALE"
+
+
+@dataclass(frozen=True)
+class AccountableFeedConfig:
+    """Binds a feed id to the dashboard that serves it.
+
+    Args:
+        dfid: Accountable data feed id, e.g. ``"100000026"``.
+        dashboard_url: Public JSON endpoint for the report.
+        message_url: Public URL for the dashboard, used in alerts.
+        dashboard_type: Dashboard type the endpoint serves
+        required_sources: Source names that must carry usable freshness metadata.
+        source_frequency_overrides: Trusted source-frequency corrections keyed
+            by source name, used when the JSON endpoint disagrees with the UI.
+    """
+
+    dfid: str
+    dashboard_url: str
+    message_url: str
+    dashboard_type: str
+    required_sources: tuple[str, ...] = ()
+    source_frequency_overrides: tuple[tuple[str, str], ...] = ()
+
+
+@dataclass(frozen=True)
+class DataSourceSnapshot:
+    """One upstream data source feeding a Proof of Solvency report."""
+
+    name: str
+    source_type: str
+    frequency: str
+    last_updated_ms: int
+    age_seconds: int
+    max_age_seconds: int
+
+    @property
+    def is_stale(self) -> bool:
+        """Whether the source is older than its cadence-based age limit."""
+        return self.age_seconds > self.max_age_seconds
+
+
+@dataclass(frozen=True)
+class AccountableReport:
+    """A validated Proof of Solvency report.
+
+    ``collateralization`` is recomputed from ``total_reserves / total_supply``
+    at full precision. ``reported_collateralization`` is the (rounded) value the
+    API returned, retained for cross-checking and display.
+
+    ``source_problems`` describes required sources whose freshness could not be
+    established. The ratio is still trustworthy in that case, so the report is
+    returned as ``STALE`` rather than withheld.
+    """
+
+    dfid: str
+    collateralization: Decimal
+    reported_collateralization: Decimal
+    net: Decimal
+    total_reserves: Decimal
+    total_supply: Decimal
+    verifiability: Decimal
+    ts_ms: int
+    report_age_seconds: int
+    report_interval: str
+    report_cadence_seconds: int
+    sources: tuple[DataSourceSnapshot, ...]
+    source_problems: tuple[str, ...] = ()
+
+    @property
+    def stale_sources(self) -> tuple[DataSourceSnapshot, ...]:
+        """Sources older than their cadence-based age limits."""
+        return tuple(source for source in self.sources if source.is_stale)
+
+    @property
+    def report_is_stale(self) -> bool:
+        """Whether the aggregate report is older than its cadence-based limit."""
+        return self.report_age_seconds > _stale_after_seconds(self.report_cadence_seconds)
+
+    @property
+    def report_timestamp(self) -> datetime:
+        """Report timestamp as a timezone-aware UTC datetime."""
+        return datetime.fromtimestamp(self.ts_ms / MS_PER_SECOND, tz=timezone.utc)
+
+
+@dataclass(frozen=True)
+class AccountableFetchResult:
+    """Result of fetching a feed: always a status, never a silent ``None``.
+
+    A ``STALE`` result still carries its report so callers can log the values
+    while alerting on the staleness.
+    """
+
+    status: AccountableStatus
+    report: AccountableReport | None
+    reason: str = ""
+
+    @property
+    def is_ok(self) -> bool:
+        return self.status is AccountableStatus.OK
+
+
+def _coerce_decimal(value: Any, field_name: str) -> Decimal:
+    """Convert an API value to Decimal, accepting numeric strings.
+
+    The live payload types several documented numbers as strings (``ts``,
+    ``verifiability``, ``timeline[].point``), so strict type checks would reject
+    valid data.
+
+    Raises:
+        AccountableError: If the value is missing, non-numeric, or not finite.
+    """
+    if isinstance(value, bool) or value is None:
+        raise AccountableError(f"{field_name} is not numeric: {value!r}")
+    if isinstance(value, float) and not math.isfinite(value):
+        raise AccountableError(f"{field_name} is not finite: {value!r}")
+    try:
+        result = Decimal(str(value).strip())
+    except (InvalidOperation, ValueError, TypeError) as exc:
+        raise AccountableError(f"{field_name} is not numeric: {value!r}") from exc
+    if not result.is_finite():
+        raise AccountableError(f"{field_name} is not finite: {value!r}")
+    return result
+
+
+def _coerce_int(value: Any, field_name: str) -> int:
+    """Convert an API value to int, accepting numeric strings."""
+    return int(_coerce_decimal(value, field_name))
+
+
+def _require_mapping(payload: Any, field_name: str) -> dict[str, Any]:
+    """Return ``payload`` as a dict or raise."""
+    if not isinstance(payload, dict):
+        raise AccountableError(f"{field_name} is missing or not an object")
+    return payload
+
+
+def _require_value(container: dict[str, Any], key: str) -> Decimal:
+    """Read ``container[key]['value']`` as a Decimal."""
+    entry = _require_mapping(container.get(key), key)
+    if "value" not in entry:
+        raise AccountableError(f"{key}.value is missing")
+    return _coerce_decimal(entry["value"], f"{key}.value")
+
+
+def parse_frequency_seconds(frequency: Any) -> int | None:
+    """Parse a declared source cadence into seconds.
+
+    Handles named cadences (``"DAILY"``, ``"WEEKLY"``, ``"live"``) and
+    quantity/unit forms (``"15 MIN"``, ``"48 H"``, ``"1 W"``).
+
+    Returns:
+        Cadence in seconds, or None when the format is unrecognised.
+    """
+    if not isinstance(frequency, str):
+        return None
+    normalized = frequency.strip().upper()
+    if not normalized:
+        return None
+    if normalized in _NAMED_FREQUENCIES:
+        return _NAMED_FREQUENCIES[normalized]
+
+    parts = normalized.split()
+    if len(parts) == 2:
+        quantity_text, unit = parts
+    else:
+        # Accept compact forms such as "48H".
+        digits = "".join(ch for ch in normalized if ch.isdigit())
+        unit = normalized[len(digits) :].strip()
+        quantity_text = digits
+    if not quantity_text or unit not in _FREQUENCY_UNIT_SECONDS:
+        return None
+    try:
+        quantity = int(quantity_text)
+    except ValueError:
+        return None
+    if quantity <= 0:
+        return None
+    return quantity * _FREQUENCY_UNIT_SECONDS[unit]
+
+
+def _parse_data_sources(
+    payload: Any,
+    now_ms: int,
+    required_sources: tuple[str, ...] = (),
+    frequency_overrides: tuple[tuple[str, str], ...] = (),
+) -> tuple[tuple[DataSourceSnapshot, ...], tuple[str, ...]]:
+    """Build source snapshots with cadence-based staleness budgets.
+
+    Unknown sources with an unrecognised cadence or missing timestamp are
+    skipped, so an Accountable schema addition cannot spuriously page us. A
+    configured required source that is missing or unparseable is recorded as a
+    problem rather than raised: freshness can no longer be established, but the
+    collateral ratio itself is unaffected and must still be evaluated. A source
+    rename upstream degrades the feed to ``STALE``, it does not blind the
+    collateral-ratio check.
+
+    Returns:
+        The parsed snapshots, and descriptions of any required-source problems.
+    """
+    if not isinstance(payload, dict):
+        if required_sources:
+            return (), ("dataSources is missing or not an object",)
+        return (), ()
+
+    required = set(required_sources)
+    overrides = dict(frequency_overrides)
+    problems: list[str] = []
+    missing = sorted(required.difference(payload))
+    if missing:
+        problems.append(f"required dataSources are missing: {', '.join(missing)}")
+
+    snapshots: list[DataSourceSnapshot] = []
+    for name, entry in payload.items():
+        is_required = name in required
+        if not isinstance(entry, dict):
+            if is_required:
+                problems.append(f"dataSources.{name} is not an object")
+            continue
+        reported_frequency = entry.get("frequency")
+        effective_frequency = overrides.get(str(name), reported_frequency)
+        cadence_seconds = parse_frequency_seconds(effective_frequency)
+        if cadence_seconds is None:
+            if is_required:
+                problems.append(f"dataSources.{name}.frequency is not recognised: {effective_frequency!r}")
+            else:
+                logger.debug("Accountable source %s has unparseable frequency %r", name, effective_frequency)
+            continue
+        if str(name) in overrides and reported_frequency != effective_frequency:
+            logger.warning(
+                "Accountable source %s reports frequency %r; using configured override %r",
+                name,
+                reported_frequency,
+                effective_frequency,
+            )
+        try:
+            last_updated_ms = _coerce_int(entry.get("lastUpdated"), f"dataSources.{name}.lastUpdated")
+        except AccountableError as exc:
+            if is_required:
+                problems.append(str(exc))
+            else:
+                logger.debug("Accountable source %s has no usable lastUpdated", name)
+            continue
+
+        age_seconds = (now_ms - last_updated_ms) // MS_PER_SECOND
+        if age_seconds < -MAX_FUTURE_SKEW_SECONDS:
+            # Clamping a future timestamp to age 0 would make a source with a
+            # broken clock look permanently fresh, which is the one thing the
+            # freshness check exists to catch.
+            if is_required:
+                problems.append(f"dataSources.{name}.lastUpdated is {-age_seconds}s in the future")
+            else:
+                logger.debug("Accountable source %s has a future lastUpdated", name)
+            continue
+
+        source_type_value = entry.get("type")
+        if not isinstance(source_type_value, str) or not source_type_value.strip():
+            if is_required:
+                problems.append(f"dataSources.{name}.type is missing or not a string")
+                continue
+            source_type = ""
+        else:
+            source_type = source_type_value
+        snapshots.append(
+            DataSourceSnapshot(
+                name=str(name),
+                source_type=source_type,
+                frequency=str(effective_frequency),
+                last_updated_ms=last_updated_ms,
+                age_seconds=max(0, age_seconds),
+                max_age_seconds=_stale_after_seconds(cadence_seconds),
+            )
+        )
+    return tuple(snapshots), tuple(problems)
+
+
+def _validate_consistency(
+    collateralization: Decimal,
+    reported: Decimal,
+    net: Decimal,
+    total_reserves: Decimal,
+    total_supply: Decimal,
+) -> None:
+    """Cross-check the reported ratio and net against the underlying totals.
+
+    Raises:
+        AccountableError: If either identity fails its tolerance.
+    """
+    if abs(collateralization - reported) > RATIO_CONSISTENCY_TOLERANCE:
+        raise AccountableError(
+            f"collateralization {reported} disagrees with total_reserves/total_supply {collateralization}"
+        )
+
+    net_tolerance = max(
+        NET_CONSISTENCY_ABSOLUTE_TOLERANCE,
+        NET_CONSISTENCY_RELATIVE_TOLERANCE * abs(total_reserves),
+    )
+    expected_net = total_reserves - total_supply
+    if abs(net - expected_net) > net_tolerance:
+        raise AccountableError(f"net {net} disagrees with total_reserves - total_supply {expected_net}")
+
+
+def _validate_usd_supply(supply_entry: dict[str, Any]) -> None:
+    """Establish that raw supply is the USD liability denominator.
+
+    Accountable documents ``total_supply.fx`` as 1 for USD-pegged feeds, but the
+    live 3Jane response currently omits the field. When present, require it to be
+    exactly 1.
+
+    When absent, the invariant is already enforced by the net cross-check in
+    :func:`_validate_consistency`: the server computes ``net`` against
+    liabilities, so ``net ≈ total_reserves - total_supply`` holds only when
+    liabilities equal raw supply, i.e. when fx is 1. A non-pegged feed fails
+    there instead, on the same tolerance.
+    """
+    if "fx" not in supply_entry:
+        return
+    fx = _coerce_decimal(supply_entry["fx"], "total_supply.fx")
+    if fx != 1:
+        raise AccountableError(f"total_supply.fx is {fx}, expected 1 (non-USD-pegged feed is unsupported)")
+
+
+def parse_report(payload: Any, config: AccountableFeedConfig, now_ms: int) -> AccountableReport:
+    """Validate a raw dashboard payload into an :class:`AccountableReport`.
+
+    Args:
+        payload: Decoded JSON body from the dashboard endpoint.
+        config: Feed configuration supplying the DFID to bind the report to.
+        now_ms: Current time in milliseconds, for age calculations.
+
+    Raises:
+        AccountableError: On any schema, type, range, or consistency violation.
+    """
+    body = _require_mapping(payload, "response")
+    if body.get("res") != "ok":
+        raise AccountableError(f"response res is {body.get('res')!r}, expected 'ok'")
+
+    data = _require_mapping(body.get("data"), "data")
+    reserves = _require_mapping(data.get("reserves"), "data.reserves")
+
+    total_reserves = _require_value(reserves, "total_reserves")
+    total_supply = _require_value(reserves, "total_supply")
+    if total_reserves < MIN_PLAUSIBLE_TOTAL or total_supply < MIN_PLAUSIBLE_TOTAL:
+        raise AccountableError(f"implausible totals: reserves={total_reserves}, supply={total_supply}")
+
+    supply_entry = _require_mapping(reserves.get("total_supply"), "total_supply")
+    reported_collateralization = _coerce_decimal(data.get("collateralization"), "collateralization")
+    net = _coerce_decimal(data.get("net"), "net")
+
+    # The docs define collateralization and net against liabilities, which equal
+    # raw total_supply only for a USD-pegged feed. Validate the explicit fx when
+    # available; the net cross-check below enforces the same invariant when it
+    # is absent, as it is on the live 3Jane response.
+    _validate_usd_supply(supply_entry)
+
+    collateralization = total_reserves / total_supply
+    if not MIN_PLAUSIBLE_RATIO <= collateralization <= MAX_PLAUSIBLE_RATIO:
+        raise AccountableError(f"implausible collateralization ratio: {collateralization}")
+
+    _validate_consistency(collateralization, reported_collateralization, net, total_reserves, total_supply)
+
+    ts_ms = _coerce_int(data.get("ts"), "ts")
+    age_seconds = (now_ms - ts_ms) // MS_PER_SECOND
+    if age_seconds < -MAX_FUTURE_SKEW_SECONDS:
+        raise AccountableError(f"report timestamp is {-age_seconds}s in the future")
+
+    report_interval_value = reserves.get("interval")
+    report_cadence_seconds = parse_frequency_seconds(report_interval_value)
+    if report_cadence_seconds is None:
+        raise AccountableError(f"reserves.interval is not recognised: {report_interval_value!r}")
+
+    sources, source_problems = _parse_data_sources(
+        data.get("dataSources"),
+        now_ms,
+        config.required_sources,
+        config.source_frequency_overrides,
+    )
+
+    return AccountableReport(
+        dfid=config.dfid,
+        collateralization=collateralization,
+        reported_collateralization=reported_collateralization,
+        net=net,
+        total_reserves=total_reserves,
+        total_supply=total_supply,
+        verifiability=_coerce_decimal(reserves.get("verifiability"), "reserves.verifiability"),
+        ts_ms=ts_ms,
+        report_age_seconds=max(0, age_seconds),
+        report_interval=str(report_interval_value),
+        report_cadence_seconds=report_cadence_seconds,
+        sources=sources,
+        source_problems=source_problems,
+    )
+
+
+def evaluate_report(report: AccountableReport) -> AccountableFetchResult:
+    """Classify a parsed report as OK or STALE.
+
+    Short cadences of up to one hour become stale after two periods. Longer
+    cadences become stale after the first missed update. Required sources whose
+    freshness cannot be established are also stale.
+    """
+    if report.report_is_stale:
+        return AccountableFetchResult(
+            AccountableStatus.STALE,
+            report,
+            f"report is {report.report_age_seconds // SECONDS_PER_HOUR}h old (interval {report.report_interval})",
+        )
+
+    if report.source_problems:
+        return AccountableFetchResult(
+            AccountableStatus.STALE,
+            report,
+            f"unusable source freshness metadata: {'; '.join(report.source_problems)}",
+        )
+
+    stale = report.stale_sources
+    if stale:
+        detail = ", ".join(
+            f"{source.name} ({source.age_seconds // SECONDS_PER_HOUR}h old, cadence {source.frequency})"
+            for source in stale
+        )
+        return AccountableFetchResult(AccountableStatus.STALE, report, f"stale sources: {detail}")
+
+    return AccountableFetchResult(AccountableStatus.OK, report)
+
+
+def fetch_report(config: AccountableFeedConfig, now_ms: int | None = None) -> AccountableFetchResult:
+    """Fetch, validate, and classify a Proof of Solvency report.
+
+    Network and schema failures are converted into an ``UNAVAILABLE`` result
+    rather than raised, so a feed outage cannot interrupt a caller's other
+    checks. Retries follow ``utils.http_client`` for rate limiting, transient
+    5xx responses, connection errors, and timeouts.
+
+    Args:
+        config: Feed to fetch.
+        now_ms: Current time in milliseconds; defaults to wall clock.
+
+    Returns:
+        A result carrying the status and, when parseable, the report.
+    """
+    if now_ms is None:
+        now_ms = int(datetime.now(tz=timezone.utc).timestamp() * MS_PER_SECOND)
+
+    # The URL is overridable by env, and the feed is authenticated by TLS alone
+    # in v1 (no signature verification yet), so plaintext is not acceptable.
+    if not config.dashboard_url.lower().startswith("https://"):
+        logger.error("Accountable feed %s has a non-HTTPS URL: %s", config.dfid, config.dashboard_url)
+        return AccountableFetchResult(AccountableStatus.UNAVAILABLE, None, "dashboard URL is not HTTPS")
+
+    try:
+        response = request_with_retry("get", config.dashboard_url, headers={"Accept": "application/json"})
+    except requests.RequestException as exc:
+        logger.warning("Accountable feed %s unreachable: %s", config.dfid, exc)
+        return AccountableFetchResult(AccountableStatus.UNAVAILABLE, None, f"request failed: {exc}")
+
+    # Kept out of the block above: requests raises JSONDecodeError, which is
+    # itself a RequestException, so a shared handler would report a decode
+    # failure as a network failure.
+    try:
+        payload = response.json()
+    except ValueError as exc:
+        logger.warning("Accountable feed %s returned non-JSON: %s", config.dfid, exc)
+        return AccountableFetchResult(AccountableStatus.UNAVAILABLE, None, f"invalid JSON: {exc}")
+
+    try:
+        report = parse_report(payload, config, now_ms)
+    except AccountableError as exc:
+        logger.warning("Accountable feed %s failed validation: %s", config.dfid, exc)
+        return AccountableFetchResult(AccountableStatus.UNAVAILABLE, None, str(exc))
+
+    return evaluate_report(report)
diff --git a/utils/formatting.py b/utils/formatting.py
index 159a1ca8..0c04991b 100644
--- a/utils/formatting.py
+++ b/utils/formatting.py
@@ -2,13 +2,19 @@
 
 
 def format_with_suffix(number: float) -> str:
-    """Format number with K, M, B suffixes for readability."""
-    if number >= 1_000_000_000:
-        return f"{number / 1_000_000_000:.2f}B"
-    if number >= 1_000_000:
-        return f"{number / 1_000_000:.2f}M"
-    if number >= 1_000:
-        return f"{number / 1_000:.2f}K"
+    """Format number with K, M, B suffixes for readability.
+
+    Negative values are suffixed on their magnitude, so a shortfall reads as
+    "-2.50M" rather than "-2500000.00".
+    """
+    sign = "-" if number < 0 else ""
+    magnitude = abs(number)
+    if magnitude >= 1_000_000_000:
+        return f"{sign}{magnitude / 1_000_000_000:.2f}B"
+    if magnitude >= 1_000_000:
+        return f"{sign}{magnitude / 1_000_000:.2f}M"
+    if magnitude >= 1_000:
+        return f"{sign}{magnitude / 1_000:.2f}K"
     return f"{number:.2f}"
 
 
diff --git a/utils/http_client.py b/utils/http_client.py
index de5209f1..b4637a98 100644
--- a/utils/http_client.py
+++ b/utils/http_client.py
@@ -21,8 +21,8 @@ def request_with_retry(
 ) -> requests.Response:
     """Make an HTTP request with exponential backoff retry on transient errors.
 
-    Retries on 5xx server errors, connection errors, and timeouts.
-    Raises immediately on 4xx client errors.
+    Retries on rate limiting (429), 5xx server errors, connection errors, and
+    timeouts. Raises immediately on other 4xx client errors.
 
     Args:
         method: HTTP method (get, post, etc.).
@@ -53,8 +53,8 @@ def request_with_retry(
             return response
         except requests.exceptions.HTTPError as e:
             status_code = e.response.status_code if e.response is not None else None
-            if status_code is not None and status_code < 500:
-                raise  # Don't retry client errors (4xx)
+            if status_code is not None and status_code < 500 and status_code != 429:
+                raise  # Do not retry permanent client errors.
             last_exception = e
         except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as e:
             last_exception = e