Prioritize latest external signals

Author: WW-shan

poly_strategy/external_signals.py

@@ -70,7 +70,9 @@ def polymarket_market_ids_from_external_signals(path: Path, limit: Optional[int]
         raise ValueError("limit must be at least 1")
     market_ids = []
     seen = set()
-    for row in _read_external_signals(path):
+    rows = list(_read_external_signals(path))
+    iterable = reversed(rows) if limit is not None else rows
+    for row in iterable:
         for leg in row.get("legs", []):
             if str(leg.get("venue") or "").lower() != "polymarket":
                 continue

poly_strategy/watchlist.py

@@ -264,30 +264,34 @@ def _top_neg_risk_group_market_ids(
 
 
 def _external_signal_market_scores(path: Optional[Path], alias_to_market_id: Optional[dict] = None) -> Counter:
-    scores = Counter()
+    scores = {}
     if not path or not path.exists():
-        return scores
+        return Counter()
     alias_to_market_id = alias_to_market_id or {}
+    rows = []
     with path.open() as handle:
         for raw_line in handle:
             line = raw_line.strip()
             if not line:
                 continue
             try:
-                row = json.loads(line)
+                rows.append(json.loads(line))
             except json.JSONDecodeError:
                 continue
-            if row.get("type") != "external_signal":
+    for index, row in enumerate(rows):
+        signal_score = 100_000.0 + 10_000.0 * float(row.get("quoted_edge") or 0.0) + float(index)
+        if row.get("type") != "external_signal":
+            continue
+        for leg in row.get("legs", []):
+            if str(leg.get("venue") or "").lower() != "polymarket":
                 continue
-            signal_score = 100_000.0 + 10_000.0 * float(row.get("quoted_edge") or 0.0)
-            for leg in row.get("legs", []):
-                if str(leg.get("venue") or "").lower() != "polymarket":
-                    continue
-                raw_market_id = str(leg.get("market_id") or "").strip()
-                market_id = alias_to_market_id.get(raw_market_id, raw_market_id)
-                if market_id:
-                    scores[market_id] += signal_score
-    return scores
+            raw_market_id = str(leg.get("market_id") or "").strip()
+            market_id = alias_to_market_id.get(raw_market_id, raw_market_id)
+            if market_id:
+                current = scores.get(market_id)
+                if current is None or signal_score > current:
+                    scores[market_id] = signal_score
+    return Counter(scores)
 
 
 def _canonicalize_market_ids(market_ids: Iterable[str], alias_to_market_id: dict) -> set:

tests/test_external_signals.py

@@ -213,6 +213,23 @@ def test_polymarket_market_ids_from_external_signals_dedupes_in_order(self):
 
         self.assertEqual(market_ids, ["pm-1", "pm-2"])
 
+    def test_polymarket_market_ids_from_external_signals_limit_prefers_latest_unique_ids(self):
+        rows = [
+            {"type": "external_signal", "legs": [{"venue": "polymarket", "market_id": "old-1"}]},
+            {"type": "external_signal", "legs": [{"venue": "polymarket", "market_id": "old-2"}]},
+            {"type": "external_signal", "legs": [{"venue": "polymarket", "market_id": "new-1"}]},
+            {"type": "external_signal", "legs": [{"venue": "polymarket", "market_id": "old-1"}]},
+            {"type": "external_signal", "legs": [{"venue": "polymarket", "market_id": "new-2"}]},
+        ]
+
+        with tempfile.TemporaryDirectory() as tmp:
+            path = Path(tmp) / "external.ndjson"
+            path.write_text("\n".join(json.dumps(row) for row in rows) + "\n")
+
+            market_ids = polymarket_market_ids_from_external_signals(path, limit=3)
+
+        self.assertEqual(market_ids, ["new-2", "old-1", "new-1"])
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_watchlist.py

@@ -204,6 +204,41 @@ def test_build_polymarket_watchlist_keeps_external_signals_under_cap(self):
         signal = next(row for row in rows if row["market_id"] == "signal")
         self.assertIn("external_signal", signal["priority_reasons"])
 
+    def test_build_polymarket_watchlist_cap_prefers_latest_external_signal(self):
+        gamma_rows = [
+            _gamma_row("old-signal", "", ["old-yes", "old-no"], "", liquidity=0, volume24hr=0),
+            _gamma_row("new-signal", "", ["new-yes", "new-no"], "", liquidity=0, volume24hr=0),
+            _gamma_row("top", "", ["top-yes", "top-no"], "", liquidity=1000, volume24hr=1000),
+        ]
+        rules = {"mutually_exclusive": []}
+        old_signal = {
+            "type": "external_signal",
+            "legs": [{"venue": "polymarket", "market_id": "old-signal"}],
+        }
+        new_signal = {
+            "type": "external_signal",
+            "legs": [{"venue": "polymarket", "market_id": "new-signal"}],
+        }
+
+        with tempfile.TemporaryDirectory() as tmp:
+            gamma_path = Path(tmp) / "gamma.ndjson"
+            rules_path = Path(tmp) / "rules.json"
+            signals_path = Path(tmp) / "signals.ndjson"
+            gamma_path.write_text("\n".join(json.dumps(row) for row in gamma_rows) + "\n")
+            rules_path.write_text(json.dumps(rules))
+            signals_path.write_text("\n".join(json.dumps(row) for row in [old_signal, old_signal, old_signal, new_signal]) + "\n")
+
+            rows = build_polymarket_watchlist(
+                gamma_path,
+                rules_path,
+                external_signals_path=signals_path,
+                include_top_markets=1,
+                max_markets=1,
+            )
+
+        self.assertIn("new-signal", {row["market_id"] for row in rows})
+        self.assertNotIn("old-signal", {row["market_id"] for row in rows})
+
     def test_build_polymarket_watchlist_keeps_external_signal_neg_risk_group_atomic(self):
         gamma_rows = [
             _gamma_row("group-a", "group", ["ga-yes", "ga-no"], "1", liquidity=100, volume24hr=10),