Add focused maker no-fill diagnostics

Author: WW-shan

docs/command-reference.md

@@ -500,6 +500,30 @@ Run a realtime-specific analysis report explaining why opportunities are absent
 The `zero_opportunity_diagnosis` section separates actionable near-misses from diagnostic or blocked candidates, so a positive-looking basket that still needs rule promotion will not be treated as executable.
 The same report is refreshed by `scripts/run_realtime_analysis_once.sh`; the LaunchAgent `poly_strategy_realtime_analysis_15m` runs it every 15 minutes.
 
+Extract markets from a specific optimization lever and run a focused maker-fee scan:
+
+```bash
+.venv/bin/python -m poly_strategy.cli optimization-target-markets data/realtime-monitor-24h-v1-analysis.json \
+  --lever maker_fee_avoidance \
+  --top-targets 1 \
+  --max-markets 120 \
+  --out data/optimization-target-market-ids.txt
+
+MAX_TARGET_MARKETS=120 TOP=50 scripts/run_maker_focus_from_analysis_once.sh
+```
+
+Validate the focused maker candidates against public SELL trade prints:
+
+```bash
+HYBRID_SCAN=data/maker-hybrid-scan-focus.json \
+TRADES=data/polymarket-data-trades-focus.ndjson \
+OUT=data/maker-hybrid-tape-sim-focus.json \
+LOCK_DIR=var/locks/maker-hybrid-tape-focus.lock \
+scripts/run_maker_hybrid_tape_once.sh
+```
+
+`maker-hybrid-tape-sim` remains diagnostic-only, but its report explains why fills did not complete with `rejection_by_reason`, `maker_fill_progress_distribution`, and `top_unfilled_maker_legs`.
+
 Promote only usable opportunities from diagnostic basket candidates:
 
 ```bash

docs/pipeline.md

@@ -49,3 +49,34 @@ The report also includes `strategy_chain_breakdown`, which applies the same logi
 - `rule_verification`: a diagnostic basket has high apparent edge but must be verified before promotion.
 - `paper_filter_debugging`: stable opportunities exist but fail paper filters; inspect rejection reasons.
 - `feed_coverage`: the watchlist has many tokens without current snapshots.
+
+## Focused Maker Fee-Avoidance Loop
+
+When the top blocker is `maker_fee_avoidance`, the next loop is deliberately narrow:
+
+1. Extract the market IDs from `optimization_targets`.
+2. Refresh only those books, optionally expanding the full `negRiskMarketID` group.
+3. Run `maker-hybrid-scan` on the focused snapshot file.
+4. Validate passive fill assumptions with `maker-hybrid-tape-sim`.
+5. Read the no-fill diagnostics before making quotes more aggressive.
+
+Useful one-shot commands:
+
+```bash
+python3 -m poly_strategy.cli optimization-target-markets \
+  data/realtime-monitor-24h-v1-analysis.json \
+  --lever maker_fee_avoidance \
+  --top-targets 1 \
+  --max-markets 120 \
+  --out data/optimization-target-market-ids.txt
+
+MAX_TARGET_MARKETS=120 TOP=50 scripts/run_maker_focus_from_analysis_once.sh
+```
+
+The tape report is diagnostic-only. It now includes:
+
+- `rejection_by_reason`: whether candidates fail because maker legs do not fill or because the hedge is no longer profitable.
+- `maker_fill_progress_distribution`: how many maker legs filled before rejection.
+- `top_unfilled_maker_legs`: repeated unfilled markets, quote levels, spread, distance to ask, and expected edge.
+
+This is the guardrail that prevents treating theoretical maker savings as a tradeable opportunity.

poly_strategy/cli.py

@@ -74,7 +74,7 @@
 )
 from poly_strategy.monitoring import IncrementalReplayState, stable_current_opportunities
 from poly_strategy.notifications import notify_alerts
-from poly_strategy.paper_analysis import analyze_paper_monitor_report
+from poly_strategy.paper_analysis import analyze_paper_monitor_report, optimization_target_market_ids
 from poly_strategy.paper import opportunity_key, select_paper_trades, trade_to_row, rejection_to_row, opportunity_to_row
 from poly_strategy.realtime import (
     DEFAULT_WS_MAX_SIZE,
@@ -152,15 +152,38 @@ def main(argv=None) -> int:
             print(f"wrote={count} out={args.out}")
             return 0
         if args.command == "collect-polymarket-binaries":
-            count = collect_polymarket_binary_snapshots_loop(
-                Path(args.out),
-                args.limit,
-                args.timeout,
-                args.proxy,
-                args.interval,
-                args.iterations,
-                max_workers=args.book_workers,
-            )
+            market_ids = list(args.market_id or [])
+            if args.market_ids_file:
+                market_ids.extend(_read_lines(Path(args.market_ids_file)))
+            if market_ids:
+                if not args.gamma:
+                    raise ValueError("--gamma is required when collecting specific market IDs")
+                count = 0
+                for index in range(args.iterations):
+                    count += collect_polymarket_binary_snapshots_for_market_ids(
+                        Path(args.out),
+                        Path(args.gamma),
+                        market_ids,
+                        args.timeout,
+                        args.proxy,
+                        max_workers=args.book_workers,
+                        skip_book_errors=args.skip_book_errors,
+                        refresh_missing_gamma=args.refresh_missing_gamma,
+                        expand_neg_risk_groups=not args.no_expand_neg_risk_groups,
+                        max_markets=args.max_markets,
+                    )
+                    if index < args.iterations - 1 and args.interval > 0:
+                        time.sleep(args.interval)
+            else:
+                count = collect_polymarket_binary_snapshots_loop(
+                    Path(args.out),
+                    args.limit,
+                    args.timeout,
+                    args.proxy,
+                    args.interval,
+                    args.iterations,
+                    max_workers=args.book_workers,
+                )
             print(f"wrote={count} out={args.out}")
             return 0
         if args.command == "collect-polymarket-trades":
@@ -345,6 +368,21 @@ def main(argv=None) -> int:
             else:
                 print(json.dumps(row, sort_keys=True))
             return 0
+        if args.command == "optimization-target-markets":
+            report = json.loads(Path(args.analysis).read_text())
+            market_ids = optimization_target_market_ids(
+                report,
+                lever=args.lever,
+                top_targets=args.top_targets,
+                max_markets=args.max_markets,
+            )
+            if args.out:
+                Path(args.out).parent.mkdir(parents=True, exist_ok=True)
+                Path(args.out).write_text("\n".join(market_ids) + ("\n" if market_ids else ""))
+                print(f"market_ids={len(market_ids)} lever={args.lever} out={args.out}")
+            else:
+                print(json.dumps({"market_ids": market_ids, "market_id_count": len(market_ids)}, sort_keys=True))
+            return 0
         if args.command == "maker-scan":
             row = maker_scan_report(
                 Path(args.snapshots),
@@ -1153,6 +1191,17 @@ def _build_parser() -> argparse.ArgumentParser:
     collect_binaries.add_argument("--iterations", type=int, default=1, help="number of collection iterations")
     collect_binaries.add_argument("--interval", type=float, default=0.0, help="seconds between iterations")
     collect_binaries.add_argument("--book-workers", type=int, default=1, help="parallel CLOB book fetch workers")
+    collect_binaries.add_argument("--gamma", help="raw Polymarket Gamma NDJSON path when collecting specific market IDs")
+    collect_binaries.add_argument("--market-id", action="append", help="Gamma market ID; can be repeated")
+    collect_binaries.add_argument("--market-ids-file", help="newline-delimited Gamma market IDs")
+    collect_binaries.add_argument("--refresh-missing-gamma", action="store_true", help="fetch missing Gamma metadata by market ID")
+    collect_binaries.add_argument("--max-markets", type=int, help="cap collected markets after optional neg-risk expansion")
+    collect_binaries.add_argument("--skip-book-errors", action="store_true", help="skip CLOB book errors instead of failing")
+    collect_binaries.add_argument(
+        "--no-expand-neg-risk-groups",
+        action="store_true",
+        help="do not expand selected markets to their full known negRiskMarketID group",
+    )
 
     collect_trades = subparsers.add_parser(
         "collect-polymarket-trades",
@@ -1336,6 +1385,16 @@ def _build_parser() -> argparse.ArgumentParser:
         help="minimum net edge threshold used to classify near misses",
     )
 
+    optimization_markets = subparsers.add_parser(
+        "optimization-target-markets",
+        help="extract market IDs from monitor-analysis optimization targets",
+    )
+    optimization_markets.add_argument("analysis", help="monitor-analysis JSON path")
+    optimization_markets.add_argument("--out", help="newline-delimited output path; prints JSON when omitted")
+    optimization_markets.add_argument("--lever", default="maker_fee_avoidance", help="target lever, or top/all")
+    optimization_markets.add_argument("--top-targets", type=int, default=1, help="number of matching targets to use")
+    optimization_markets.add_argument("--max-markets", type=int, help="cap returned market IDs")
+
     maker_scan = subparsers.add_parser(
         "maker-scan",
         help="scan latest snapshots for passive maker basket candidates without submitting orders",

poly_strategy/maker.py

@@ -1,4 +1,4 @@
-from collections import defaultdict
+from collections import Counter, defaultdict
 import json
 from datetime import datetime, timezone
 from itertools import combinations
@@ -668,6 +668,9 @@ def maker_hybrid_tape_sim_report(
         "diagnostic_only": True,
         "diagnostic_warning": "public trade prints can prove sell-through, but queue position is still uncertain without live order fills",
         "by_kind": _hybrid_fill_summary_by_kind(results),
+        "rejection_by_reason": _hybrid_rejection_summary(results),
+        "maker_fill_progress_distribution": _maker_fill_progress_distribution(results),
+        "top_unfilled_maker_legs": _top_unfilled_maker_legs(results, top_n),
         "top_completed": sorted(completed, key=_hybrid_result_sort_key)[:top_n],
         "top_unique_completed": sorted(unique_completed, key=_hybrid_result_sort_key)[:top_n],
         "top_unsafe": sorted(unsafe, key=_hybrid_result_sort_key)[:top_n],
@@ -2571,6 +2574,184 @@ def _hybrid_fill_summary_by_kind(rows: List[dict]) -> list:
     return sorted(summary.values(), key=lambda row: (-row["completed_count"], -row["max_completed_realized_edge_at_cap"], row["kind"]))
 
 
+def _hybrid_rejection_summary(rows: List[dict]) -> list:
+    summary = {}
+    for row in rows:
+        reason = str(row.get("rejection_reason") or ("completed" if row.get("completed") else "unknown"))
+        item = summary.setdefault(
+            reason,
+            {
+                "reason": reason,
+                "candidate_observation_count": 0,
+                "completed_count": 0,
+                "max_expected_edge_at_cap": 0.0,
+                "max_expected_edge_per_share": 0.0,
+                "max_realized_edge_at_cap": 0.0,
+            },
+        )
+        item["candidate_observation_count"] += 1
+        if row.get("completed"):
+            item["completed_count"] += 1
+        item["max_expected_edge_at_cap"] = max(
+            item["max_expected_edge_at_cap"],
+            float(row.get("expected_edge_at_cap") or 0.0),
+        )
+        item["max_expected_edge_per_share"] = max(
+            item["max_expected_edge_per_share"],
+            float(row.get("expected_edge_per_share") or 0.0),
+        )
+        item["max_realized_edge_at_cap"] = max(
+            item["max_realized_edge_at_cap"],
+            float(row.get("realized_edge_at_cap") or 0.0),
+        )
+    return sorted(
+        summary.values(),
+        key=lambda row: (
+            -row["candidate_observation_count"],
+            -row["max_expected_edge_at_cap"],
+            row["reason"],
+        ),
+    )
+
+
+def _maker_fill_progress_distribution(rows: List[dict]) -> list:
+    counts = Counter(
+        (
+            int(row.get("filled_maker_leg_count") or 0),
+            int(row.get("maker_leg_count") or 0),
+        )
+        for row in rows
+    )
+    distribution = []
+    for (filled_count, maker_count), count in counts.items():
+        distribution.append(
+            {
+                "filled_maker_leg_count": filled_count,
+                "maker_leg_count": maker_count,
+                "candidate_observation_count": count,
+                "maker_leg_fill_ratio": filled_count / maker_count if maker_count else 0.0,
+            }
+        )
+    return sorted(
+        distribution,
+        key=lambda row: (
+            row["filled_maker_leg_count"],
+            row["maker_leg_count"],
+            -row["candidate_observation_count"],
+        ),
+    )
+
+
+def _top_unfilled_maker_legs(rows: List[dict], top_n: int) -> list:
+    if top_n <= 0:
+        return []
+    summary = {}
+    for row in rows:
+        unfilled_indices = set(int(index) for index in (row.get("unfilled_maker_indices") or []))
+        for index, leg in enumerate(row.get("maker_legs") or []):
+            source_index = _source_leg_index(leg, index)
+            key = (
+                str(leg.get("venue") or ""),
+                str(leg.get("market_id") or ""),
+                str(leg.get("token") or ""),
+                str(leg.get("token_id") or ""),
+                str(leg.get("side") or ""),
+                float(leg.get("limit_price") or 0.0),
+                str(leg.get("quote_mode") or ""),
+                int(leg.get("quote_offset_ticks") or 0),
+            )
+            item = summary.setdefault(
+                key,
+                {
+                    "venue": key[0],
+                    "market_id": key[1],
+                    "token": key[2],
+                    "token_id": key[3],
+                    "side": key[4],
+                    "limit_price": key[5],
+                    "quote_mode": key[6],
+                    "quote_offset_ticks": key[7],
+                    "best_bid": _float_or_none(leg.get("best_bid")),
+                    "best_ask": _float_or_none(leg.get("best_ask")),
+                    "spread": _float_or_none(leg.get("spread")),
+                    "candidate_observation_count": 0,
+                    "unfilled_count": 0,
+                    "max_expected_edge_at_cap": 0.0,
+                    "max_expected_edge_per_share": 0.0,
+                    "min_distance_to_best_ask": None,
+                    "max_improvement_over_best_bid": 0.0,
+                },
+            )
+            item["candidate_observation_count"] += 1
+            item["max_expected_edge_at_cap"] = max(
+                item["max_expected_edge_at_cap"],
+                float(row.get("expected_edge_at_cap") or 0.0),
+            )
+            item["max_expected_edge_per_share"] = max(
+                item["max_expected_edge_per_share"],
+                float(row.get("expected_edge_per_share") or 0.0),
+            )
+            distance = _leg_distance_to_best_ask(leg)
+            if distance is not None:
+                item["min_distance_to_best_ask"] = (
+                    distance
+                    if item["min_distance_to_best_ask"] is None
+                    else min(item["min_distance_to_best_ask"], distance)
+                )
+            improvement = _leg_improvement_over_best_bid(leg)
+            if improvement is not None:
+                item["max_improvement_over_best_bid"] = max(item["max_improvement_over_best_bid"], improvement)
+            if source_index in unfilled_indices:
+                item["unfilled_count"] += 1
+
+    rows = []
+    for item in summary.values():
+        if item["unfilled_count"] <= 0:
+            continue
+        count = item["candidate_observation_count"]
+        item["unfilled_rate"] = item["unfilled_count"] / count if count else 0.0
+        rows.append(item)
+    return sorted(
+        rows,
+        key=lambda row: (
+            -row["unfilled_count"],
+            -row["unfilled_rate"],
+            -row["max_expected_edge_at_cap"],
+            row["market_id"],
+            row["token"],
+        ),
+    )[:top_n]
+
+
+def _source_leg_index(leg: dict, default: int) -> int:
+    try:
+        return int(leg.get("source_leg_index", default))
+    except (TypeError, ValueError):
+        return default
+
+
+def _leg_distance_to_best_ask(leg: dict) -> Optional[float]:
+    distance = _float_or_none(leg.get("distance_to_best_ask"))
+    if distance is not None:
+        return distance
+    best_ask = _float_or_none(leg.get("best_ask"))
+    limit_price = _float_or_none(leg.get("limit_price"))
+    if best_ask is None or limit_price is None:
+        return None
+    return best_ask - limit_price
+
+
+def _leg_improvement_over_best_bid(leg: dict) -> Optional[float]:
+    improvement = _float_or_none(leg.get("improvement_over_best_bid"))
+    if improvement is not None:
+        return improvement
+    best_bid = _float_or_none(leg.get("best_bid"))
+    limit_price = _float_or_none(leg.get("limit_price"))
+    if best_bid is None or limit_price is None:
+        return None
+    return max(0.0, limit_price - best_bid)
+
+
 def _unique_tape_completed_rows(rows: List[dict]) -> List[dict]:
     deduped = {}
     for row in rows: