Score saved scans as of when they were taken (roadmap 3.6)

README.md

@@ -216,7 +216,9 @@ GreyNoise), both `explained`. The other 28 malicious IPs: AUC 1.000, detection 1
 With it on, both scanners read as `veneer` instead (see the note below).
 
 Raw OTX responses are saved, so OTX settings (`OTX_*` in `paradox/sources.py`) and the scoring weights
-can be re-tested offline with `--rescore`, with no API calls. `--refresh <beacon ...>` re-runs
+can be re-tested offline with `--rescore`, with no API calls. Each scan is aged as of when it was taken, so a
+rescore gives the same result on any day. Scans saved before that was recorded are listed in the report and age
+against today unless `--as-of YYYY-MM-DD` says otherwise. `--refresh <beacon ...>` re-runs
 just those beacons on the saved scans, keeping the old reading wherever a call fails. If OTX starts rate-limiting, the run
 stops early and fills in the gaps next time.
 

ROADMAP.md

@@ -24,7 +24,7 @@ Effort: **S** is about a day, **M** a few days, **L** a week or more.
 
 Each item has a permanent story ID (`PDX-n`) and a "done when" line for the
 [Muster board](https://github.com/Castor-o7/muster/blob/main/docs/board.md). A checked box means shipped, not verified.
-The watchman keeps each story's state. **Progress: 21 of 40 stories shipped (52%),** counting each story the same
+The watchman keeps each story's state. **Progress: 22 of 40 stories shipped (55%),** counting each story the same
 whatever its size, and the total grows as work finds new stories.
 
 ---
@@ -336,9 +336,9 @@ with no network calls. "Guarantee" tests must keep holding, and "known weakness"
 *Limits:* these pin today's rules against attacks we thought of, on fixtures where the other beacons are set by hand.
 They don't show what a real attacker can do, whether OTX would catch sock puppets, or how often this happens.
 
-### 3.6 Rescores that reproduce · S
+### 3.6 Rescores that reproduce · S · ✅ shipped
 
-- [ ] **PDX-28: Score saved scans as of when they were taken.** Done when: `paradox calibrate --rescore` ages each saved scan against its recorded scan time, scans without one say so in the report, and a test shows the same saved scan gets the same score on different days.
+- [x] **PDX-28: Score saved scans as of when they were taken.** Done when: `paradox calibrate --rescore` ages each saved scan against its recorded scan time, scans without one say so in the report, and a test shows the same saved scan gets the same score on different days.
 
 Saved scans are aged against today's date, so recency rules such as 1.6's window and the veneer's freshness check
 give different results depending on the day of the rescore, and a past result can't be reproduced.
@@ -589,7 +589,7 @@ and fix the top issues. Repeat after 5.6. It also doubles as class material.
 | When | Items | Why first |
 |---|---|---|
 | Now | ~~1.1, 1.2, 1.5, 3.2, 5.1~~ (done) | Small, keyless, and fix the biggest measured weakness (the scanner rule's dependence on GreyNoise); 5.1 makes that dependence visible |
-| Next | 1.3, 3.1, 3.6, 3.8, 3.9, 3.10, 4.1, ~~3.5, 4.6, 4.7, 4.8, 4.9, 4.10, 4.5, 1.7, 1.8, 3.7, 1.4, 1.6, 5.2, 5.4~~ (done) | Close the known false positives, build adversarial and hiding-place test sets, test the AI summarizer under attack, and show watch state and claim checks in the UI |
+| Next | 1.3, 3.1, 3.8, 3.9, 3.10, 4.1, ~~3.6, 3.5, 4.6, 4.7, 4.8, 4.9, 4.10, 4.5, 1.7, 1.8, 3.7, 1.4, 1.6, 5.2, 5.4~~ (done) | Close the known false positives, build adversarial and hiding-place test sets, test the AI summarizer under attack, and show watch state and claim checks in the UI |
 | Then | 2.1, 2.3, 3.3, 3.4, 4.2, ~~4.3, 4.4~~ (done), 5.3, 5.5–5.7 | New unmasking signals plus self-measurement once history has accumulated, with the UI restructured to hold them |
 | Later | 2.2, 2.4, 2.5 | Bigger extensions: domains, campaigns, more beacons |
 

paradox/calibrate.py

@@ -36,9 +36,10 @@
 import time
 
 from concurrent.futures import ThreadPoolExecutor
+from datetime import datetime, timezone
 
 from . import config, scoring
-from .sources import (BEACONS, NEIGHBOR_LISTS, Signal, _drop_networks, _feodo_index, _ip_list,
+from .sources import (BEACONS, NEIGHBOR_LISTS, Signal, _drop_networks, _feodo_index, _ip_list, parse_dt,
                       apply_otx, fetch_otx, run_beacon, virustotal_confidence)
 
 DIR = config.DATA_DIR / "calibration"
@@ -103,16 +104,17 @@ def scan_raw(ip: str) -> dict:
     others = [b for b in BEACONS if b.id != "otx"]
     with ThreadPoolExecutor(max_workers=len(others)) as pool:
         signals = list(pool.map(lambda b: run_beacon(b, ip), others))
-    return {"signals": [s.to_dict() for s in signals], "otx_raw": otx_raw, "otx_error": otx_error}
+    return {"scanned_at": datetime.now(timezone.utc).isoformat(),
+            "signals": [s.to_dict() for s in signals], "otx_raw": otx_raw, "otx_error": otx_error}
 
 
-def build_signals(ip: str, saved: dict) -> list[Signal]:
+def build_signals(ip: str, saved: dict, as_of: datetime | None = None) -> list[Signal]:
     """Rebuild a scan's signals, re-scoring OTX from its raw response with the
-    current settings in sources.py."""
+    current settings in sources.py, aged to as_of."""
     otx = next(b for b in BEACONS if b.id == "otx")
     sig = Signal(source=otx.id, name=otx.name, tier=otx.tier)
     if saved.get("otx_raw") is not None:
-        apply_otx(ip, sig, saved["otx_raw"])
+        apply_otx(ip, sig, saved["otx_raw"], as_of)
     else:
         sig.status, sig.summary = "error", saved.get("otx_error") or "no OTX data"
     others = [Signal(**s) for s in saved["signals"]]
@@ -126,9 +128,14 @@ def build_signals(ip: str, saved: dict) -> list[Signal]:
     return [sig] + others
 
 
-def collect(ips: list[str], pause: float, offline: bool) -> dict[str, list[Signal]]:
+def collect(ips: list[str], pause: float, offline: bool,
+            undated_as_of: datetime | None = None) -> tuple[dict[str, list[Signal]], dict[str, datetime], list[str]]:
+    """The signals for each IP, when each scan counts as taken, and which scans have no
+    recorded time. Those use undated_as_of, or are left to age against today."""
     SCANS.mkdir(parents=True, exist_ok=True)
     out = {}
+    times: dict[str, datetime] = {}
+    undated = []
     limited = 0
     for i, ip in enumerate(ips, 1):
         path = SCANS / f"{ip.replace(':', '_')}.json"
@@ -153,8 +160,14 @@ def collect(ips: list[str], pause: float, offline: bool) -> dict[str, list[Signa
             limited = 0
             path.write_text(json.dumps(saved))
             time.sleep(pause)
-        out[ip] = build_signals(ip, saved)
-    return out
+        as_of = parse_dt(saved.get("scanned_at"))
+        if as_of is None:
+            undated.append(ip)
+            as_of = undated_as_of
+        if as_of is not None:
+            times[ip] = as_of
+        out[ip] = build_signals(ip, saved, as_of)
+    return out, times, undated
 
 
 def refresh(ips: list[str], beacon_ids: list[str], pause: float) -> None:
@@ -170,6 +183,7 @@ def refresh(ips: list[str], beacon_ids: list[str], pause: float) -> None:
             new = run_beacon(b, ip)
             if new.status == "ok":   # replace the saved reading, or add a beacon newer than the scan
                 saved["signals"] = [s for s in saved["signals"] if s["source"] != b.id] + [new.to_dict()]
+                saved["scanned_at"] = datetime.now(timezone.utc).isoformat()   # its newest reading mustn't be future-dated
             else:
                 print(f"  {ip}: {b.id} {new.status}, keeping saved reading", file=sys.stderr)
         path.write_text(json.dumps(saved))
@@ -185,12 +199,12 @@ def auc(pos: list[float], neg: list[float]) -> float | None:
     return wins / (len(pos) * len(neg))
 
 
-def evaluate(labels: dict, scans: dict[str, list[Signal]]) -> dict:
+def evaluate(labels: dict, scans: dict[str, list[Signal]], times: dict[str, datetime] | None = None) -> dict:
     rows = []
     for cls in ("malicious", "benign"):
         for ip in labels[cls]:
             if ip in scans:
-                sc = scoring.compute(scans[ip])
+                sc = scoring.compute(scans[ip], now=(times or {}).get(ip))
                 scanner_by = sorted({s.name for s in scans[ip]
                                      if s.status == "ok" and s.facts.get("label_kind") == "scanner"})
                 rows.append({"ip": ip, "label": cls, "threat": sc["threat"], "paradox": sc["paradox"],
@@ -284,13 +298,22 @@ def render(ev: dict) -> str:
     return "\n".join(lines) + "\n"
 
 
+def as_of_date(text: str) -> datetime:
+    try:
+        return datetime.strptime(text, "%Y-%m-%d").replace(tzinfo=timezone.utc)
+    except ValueError:
+        raise argparse.ArgumentTypeError(f"'{text}' isn't a YYYY-MM-DD date") from None
+
+
 def main(argv: list[str] | None = None) -> int:
     ap = argparse.ArgumentParser(prog="paradox calibrate", description=__doc__.split("\n\n")[0])
     ap.add_argument("--rescore", action="store_true", help="score saved scans only; no API calls")
     ap.add_argument("--new-labels", action="store_true", help="draw a fresh labeled set")
     ap.add_argument("--size", type=int, default=30, help="malicious sample size (default 30)")
     ap.add_argument("--seed", type=int, default=7)
     ap.add_argument("--pause", type=float, default=1.5, help="seconds between live scans")
+    ap.add_argument("--as-of", type=as_of_date, metavar="YYYY-MM-DD",
+                    help="when to treat scans with no recorded time as taken (default: today, so results drift)")
     ap.add_argument("--refresh", nargs="+", metavar="BEACON", help="re-run only these beacon ids on saved scans")
     args = ap.parse_args(argv)
 
@@ -303,9 +326,15 @@ def main(argv: list[str] | None = None) -> int:
         if unknown:
             ap.error(f"unknown beacon id(s): {', '.join(sorted(unknown))} (OTX is re-scored with --rescore)")
         refresh(labels["malicious"] + labels["benign"], args.refresh, args.pause)
-    scans = collect(labels["malicious"] + labels["benign"], args.pause, offline=args.rescore or bool(args.refresh))
-    ev = evaluate(labels, scans)
+    scans, times, undated = collect(labels["malicious"] + labels["benign"], args.pause,
+                                    offline=args.rescore or bool(args.refresh), undated_as_of=args.as_of)
+    ev = evaluate(labels, scans, times)
+    ev["undated_scans"] = {"ips": undated, "as_of": args.as_of.date().isoformat() if args.as_of else "today"}
     print(render(ev))
+    if undated:
+        print(f"  {len(undated)} of {len(scans)} scans have no recorded time, so they were aged as of "
+              f"{ev['undated_scans']['as_of']}."
+              + ("" if args.as_of else " These results change with the day; use --as-of to fix them.") + "\n")
     (DIR / "report.json").write_text(json.dumps(ev, indent=1, default=str))
     return 0
 

paradox/sources.py

@@ -306,10 +306,10 @@ def otx_scan_share(data: dict) -> float | None:
     return round(scan / (scan + attack), 3) if scan + attack else None
 
 
-def apply_otx(ip: str, sig: Signal, data: dict) -> None:
+def apply_otx(ip: str, sig: Signal, data: dict, now: datetime | None = None) -> None:
     """Fill an OTX signal from a raw response (separate so calibration can
-    re-score saved responses offline)."""
-    s = score_otx(data)
+    re-score saved responses offline, as of when they were taken)."""
+    s = score_otx(data, now)
     sig.threat, sig.confidence = s["threat"], s["confidence"]
     sig.last_seen = iso_date(s["newest"])
     sig.link = f"https://otx.alienvault.com/indicator/ip/{ip}"

tests/test_scoring.py

@@ -1,3 +1,4 @@
+import json
 import tempfile
 import unittest
 from datetime import datetime, timedelta, timezone
@@ -463,6 +464,69 @@ def test_saved_scans_recover_the_reporter_count(self):
         self.assertEqual(abuse.facts["reporters"], 7)
 
 
+class RescoreAsOfTests(unittest.TestCase):
+    """PDX-28: a saved scan gets the same score whatever day it's rescored."""
+
+    PULSES = {"pulse_info": {"count": 3, "pulses": [
+        {"author": {"username": f"a{i}"}, "name": "attack campaign", "modified": "2026-10-20T00:00:00+00:00",
+         "indicator_type_counts": {"IPv4": 20}} for i in range(3)]}, "validation": []}
+
+    def rescore(self, saved, today, **kwargs):
+        """Run the real collect and evaluate with the clock set to `today`."""
+        from paradox import calibrate, sources as src
+        import paradox.scoring as sc
+
+        class Clock(datetime):
+            @classmethod
+            def now(cls, tz=None):
+                return today
+
+        with tempfile.TemporaryDirectory() as d:
+            scans = Path(d)
+            (scans / "45.1.1.1.json").write_text(json.dumps(saved))
+            with mock.patch.object(calibrate, "SCANS", scans), mock.patch.object(src, "datetime", Clock), \
+                    mock.patch.object(sc, "datetime", Clock):
+                signals, times, undated = calibrate.collect(["45.1.1.1"], 0, offline=True, **kwargs)
+                row = calibrate.evaluate({"malicious": ["45.1.1.1"], "benign": []}, signals, times)
+        return row["threat_mean"]["malicious"], row["paradox_mean"]["malicious"], undated
+
+    def saved(self, **fields):
+        abuse = {"source": "abuseipdb", "name": "a", "tier": "fast", "threat": 0.9, "confidence": 0.8,
+                 "last_seen": "2026-10-20", "facts": {"reporters": 5}, "summary": "x"}
+        return {"otx_raw": self.PULSES, "signals": [abuse], "otx_error": None, **fields}
+
+    DAY1, DAY2 = datetime(2026, 10, 21, tzinfo=timezone.utc), datetime(2027, 6, 1, tzinfo=timezone.utc)
+
+    def test_dated_scan_scores_the_same_on_any_day(self):
+        saved = self.saved(scanned_at="2026-10-21T00:00:00+00:00")
+        first, later = self.rescore(saved, self.DAY1), self.rescore(saved, self.DAY2)
+        self.assertEqual(first, later)
+        self.assertEqual(first[2], [])
+
+    def test_undated_scan_is_reported_and_drifts(self):
+        first, later = self.rescore(self.saved(), self.DAY1), self.rescore(self.saved(), self.DAY2)
+        self.assertEqual(first[2], ["45.1.1.1"])
+        self.assertNotEqual(first[:2], later[:2])
+
+    def test_as_of_fixes_an_undated_scan_without_hiding_it(self):
+        fixed = self.DAY1
+        first = self.rescore(self.saved(), self.DAY1, undated_as_of=fixed)
+        later = self.rescore(self.saved(), self.DAY2, undated_as_of=fixed)
+        self.assertEqual(first, later)
+        self.assertEqual(later[2], ["45.1.1.1"])
+
+    def test_recorded_time_wins_over_as_of(self):
+        saved = self.saved(scanned_at="2026-10-21T00:00:00+00:00")
+        self.assertEqual(self.rescore(saved, self.DAY2)[:2],
+                         self.rescore(saved, self.DAY2, undated_as_of=self.DAY2)[:2])
+
+    def test_new_scans_record_their_time(self):
+        from paradox import calibrate
+        with mock.patch.object(calibrate, "fetch_otx", return_value={}), mock.patch.object(
+                calibrate, "run_beacon", return_value=Signal("x", "x", "slow")):
+            self.assertIsNotNone(_sources.parse_dt(calibrate.scan_raw("45.1.1.1")["scanned_at"]))
+
+
 class ClaimTests(unittest.TestCase):
     def rdns(self, status="ok", **facts):
         s = sig("rdns", "context", None, 0.0, status=status)