Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
54 changes: 54 additions & 0 deletions tools/synth-overlay/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,54 @@
# Synth Overlay — Polymarket Edge Extension

Chrome extension that adds a live "fair value" layer on Polymarket market pages using Synth forecasts. Shows whether the current YES price is **underpriced**, **overpriced**, or **fair** and exposes edge strength (Strong / Moderate / No Edge).

## What it does

- **Inline badge** next to the decision point: YES Edge +X% (green), -X% (red), or Fair ±0.5% (neutral).
- **Detail card** on click: horizon (1h / 24h), confidence, last update.
- **Contextual only**: Overlay appears only when the page slug maps to a Synth-supported market (daily up/down, hourly up/down, or range). Unsupported markets show nothing.

## How it works

1. **Extension** (content script on `polymarket.com`) reads the page URL and extracts the market slug.
2. **Local API** (Flask on `http://127.0.0.1:8765`) is called with `GET /api/edge?slug=...`. The server uses `SynthClient` (mock or live) to load Polymarket comparison data.
3. **Edge logic** computes `edge_pct = (synth_prob - market_prob) * 100` and classifies signal (underpriced / fair / overpriced) and strength (strong / moderate / none) from thresholds.
4. **Overlay** is injected only when the API returns 200; 404 (unsupported market) keeps the page unchanged.

## Synth API usage

- `get_polymarket_daily()` — daily up/down (24h) Synth vs Polymarket.
- `get_polymarket_hourly()` — hourly up/down (1h).
- `get_polymarket_range()` — range brackets with synth vs polymarket probability per bracket.
- `get_prediction_percentiles(asset, horizon)` — used for confidence scoring (forecast spread) and optional bias in explanations; wired for both up/down and range.

## Run locally

1. Install: `pip install -r requirements.txt` (from repo root: `pip install -r tools/synth-overlay/requirements.txt`).
2. Start server (from repo root): `python tools/synth-overlay/server.py` (or from `tools/synth-overlay`: `python server.py`). Listens on `127.0.0.1:8765`.
3. Load extension: Chrome → Extensions → Load unpacked → select `tools/synth-overlay/extension`.
4. Open a Polymarket event/market URL whose slug matches a supported market (e.g. `bitcoin-up-or-down-on-february-26` for mock daily). The badge appears when the server is running and the slug is supported.

## Verify the overlay (before recording)

1. **Check the API** (server must be running):
```bash
curl -s "http://127.0.0.1:8765/api/edge?slug=bitcoin-up-or-down-on-february-26" | head -c 200
```
You should see JSON with `"signal"`, `"edge_pct"`, etc. If you see `"error"` or 404, the slug is not supported for the current mock/API.

2. **Open the exact URL** in Chrome (with the extension loaded from `extension/`):
- Daily (mock): `https://polymarket.com/event/bitcoin-up-or-down-on-february-26`
- Hourly (mock): `https://polymarket.com/event/bitcoin-up-or-down-february-25-6pm-et`
- The extension reads the slug from the path and calls the API. If the API returned 200 in step 1, the **badge** is appended to the page (near the first element with class containing "market" or "outcome", or `main`, or the top of the body).

3. **Interaction:**
- **Badge** = inline pill (e.g. "YES Edge +2.1%" or "Fair +0.3%") with strength (Strong/Moderate).
- **Click the badge** → toggles the **detail card** (Now 1h / By close 24h, confidence bar, optional "No trade" row).
- **Click "Details ▶"** in the card → opens the **slide-out panel** (explanation, invalidation, full confidence).

4. **If nothing appears:** Ensure (a) server is running, (b) you loaded the extension from `tools/synth-overlay/extension` (not the parent folder), (c) the address bar is exactly one of the supported URLs above. Open DevTools → Network: you should see a request to `127.0.0.1:8765/api/edge?slug=...` with status 200.

## Tests

From repo root: `python -m pytest tools/synth-overlay/tests/ -v`. Uses mock data; no API key required.
285 changes: 285 additions & 0 deletions tools/synth-overlay/analyzer.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,285 @@
"""
EdgeAnalyzer: consolidated dual-horizon edge analysis with confidence scoring
and human-readable signal explanations using Synth forecast percentiles.
"""

from dataclasses import dataclass
from typing import Literal

from edge import (
compute_edge_pct,
signal_from_edge,
strength_from_edge,
signals_conflict,
strength_from_horizons,
)


@dataclass
class HorizonEdge:
horizon: str
edge_pct: float
signal: str
synth_prob: float
market_prob: float


@dataclass
class AnalysisResult:
primary: HorizonEdge
secondary: HorizonEdge | None
strength: Literal["strong", "moderate", "none"]
confidence_score: float
no_trade: bool
explanation: str
invalidation: str


class EdgeAnalyzer:
"""Analyzes Synth vs Polymarket across horizons with percentile-based confidence."""

def __init__(
self,
daily_data: dict | None = None,
hourly_data: dict | None = None,
percentiles_1h: dict | None = None,
percentiles_24h: dict | None = None,
):
self._daily = daily_data
self._hourly = hourly_data
self._pct_1h = percentiles_1h
self._pct_24h = percentiles_24h

def _extract_edge(self, data: dict, horizon: str) -> HorizonEdge:
synth = float(data["synth_probability_up"])
market = float(data["polymarket_probability_up"])
edge_pct = compute_edge_pct(synth, market)
return HorizonEdge(
horizon=horizon,
edge_pct=edge_pct,
signal=signal_from_edge(edge_pct),
synth_prob=synth,
market_prob=market,
)

def _percentile_spread(self, pct_data: dict | None) -> float | None:
"""Relative spread (p95 - p05) / price. Returns None if data unavailable."""
if not pct_data:
return None
try:
steps = pct_data.get("forecast_future", {}).get("percentiles") or []
if not steps:
return None
last = steps[-1]
price = pct_data.get("current_price") or 1.0
if price <= 0:
return None
p95 = float(last.get("0.95", 0))
p05 = float(last.get("0.05", 0))
return abs(p95 - p05) / price
except (TypeError, KeyError, ValueError):
return None

def _directional_bias(self, pct_data: dict | None) -> float | None:
"""How much median deviates from current price: (p50 - price) / price."""
if not pct_data:
return None
try:
steps = pct_data.get("forecast_future", {}).get("percentiles") or []
if not steps:
return None
last = steps[-1]
price = pct_data.get("current_price") or 1.0
if price <= 0:
return None
p50 = float(last.get("0.5", 0))
return (p50 - price) / price
except (TypeError, KeyError, ValueError):
return None

def compute_confidence(self, spread_1h: float | None, spread_24h: float | None) -> float:
"""
Confidence score [0.0, 1.0] inversely proportional to forecast spread.
Narrow distributions = high confidence; wide = low.
"""
spreads = [s for s in (spread_1h, spread_24h) if s is not None]
if not spreads:
return 0.5
avg_spread = sum(spreads) / len(spreads)
if avg_spread <= 0.01:
return 1.0
if avg_spread >= 0.10:
return 0.1
return round(1.0 - (avg_spread - 0.01) / 0.09 * 0.9, 2)

def _build_explanation(self, edge_1h: HorizonEdge, edge_24h: HorizonEdge, confidence: float) -> str:
direction_1h = "higher" if edge_1h.edge_pct > 0 else "lower"
direction_24h = "higher" if edge_24h.edge_pct > 0 else "lower"
parts = []
if signals_conflict(edge_1h.signal, edge_24h.signal):
parts.append(
f"Synth forecasts {direction_1h} on the 1h horizon "
f"but {direction_24h} on the 24h horizon — signals conflict."
)
elif edge_1h.signal == "fair" and edge_24h.signal == "fair":
parts.append("Synth and Polymarket agree closely on both horizons.")
else:
dominant = "up" if edge_24h.edge_pct > 0 else "down"
parts.append(
f"Synth forecasts {dominant} probability {direction_24h} than "
f"Polymarket on both horizons: 1h by {abs(edge_1h.edge_pct)}pp, "
f"24h by {abs(edge_24h.edge_pct)}pp."
)
if confidence >= 0.7:
parts.append("Forecast distribution is narrow — high confidence.")
elif confidence <= 0.3:
parts.append("Forecast distribution is wide — low confidence, treat with caution.")
return " ".join(parts)

def _build_invalidation(self, edge_24h: HorizonEdge, bias_24h: float | None) -> str:
parts = []
if edge_24h.signal == "underpriced":
parts.append(
"This edge invalidates if price drops sharply, "
"pushing Synth probability below market."
)
elif edge_24h.signal == "overpriced":
parts.append(
"This edge invalidates if price rallies, "
"pushing Synth probability above market."
)
else:
parts.append("No meaningful edge to invalidate — market is fairly priced.")
if bias_24h is not None and abs(bias_24h) > 0.02:
direction = "upward" if bias_24h > 0 else "downward"
parts.append(f"Synth median shows a {direction} bias of {abs(bias_24h)*100:.1f}%.")
return " ".join(parts)

def analyze_range(
self,
selected_bracket: dict,
all_brackets: list[dict],
percentiles_24h: dict | None = None,
) -> AnalysisResult:
"""Analyze a range market bracket with context from all brackets."""
synth = float(selected_bracket.get("synth_probability", 0))
market = float(selected_bracket.get("polymarket_probability", 0))
edge_pct = compute_edge_pct(synth, market)
signal = signal_from_edge(edge_pct)
strength = strength_from_edge(edge_pct)
title = selected_bracket.get("title", "")

spread_24h = self._percentile_spread(percentiles_24h)
confidence = self.compute_confidence(None, spread_24h)
high_uncertainty = spread_24h is not None and spread_24h > 0.05
no_trade = strength == "none" or high_uncertainty

mispriced = [
b for b in all_brackets
if abs(float(b.get("synth_probability", 0)) - float(b.get("polymarket_probability", 0))) > 0.005
]
explanation = self._build_range_explanation(
title, edge_pct, signal, len(mispriced), len(all_brackets), confidence
)
invalidation = self._build_range_invalidation(
selected_bracket, signal
)

primary = HorizonEdge(
horizon="24h",
edge_pct=edge_pct,
signal=signal,
synth_prob=synth,
market_prob=market,
)
return AnalysisResult(
primary=primary,
secondary=None,
strength=strength,
confidence_score=confidence,
no_trade=no_trade,
explanation=explanation,
invalidation=invalidation,
)

def _build_range_explanation(
self,
title: str,
edge_pct: float,
signal: str,
mispriced_count: int,
total_count: int,
confidence: float,
) -> str:
parts = []
if signal == "fair":
parts.append(
f"Bracket {title}: Synth and Polymarket agree closely "
f"(edge {edge_pct:+.1f}pp)."
)
else:
direction = "higher" if edge_pct > 0 else "lower"
parts.append(
f"Bracket {title}: Synth assigns {direction} probability "
f"than Polymarket by {abs(edge_pct):.1f}pp."
)
if mispriced_count > 1:
parts.append(
f"{mispriced_count} of {total_count} brackets show mispricing."
)
if confidence >= 0.7:
parts.append("Forecast distribution is narrow — high confidence.")
elif confidence <= 0.3:
parts.append("Forecast distribution is wide — low confidence, treat with caution.")
return " ".join(parts)

def _build_range_invalidation(self, bracket: dict, signal: str) -> str:
title = bracket.get("title", "")
if signal == "underpriced":
return (
f"Edge on {title} invalidates if price moves away from this range, "
f"reducing the probability of landing here."
)
if signal == "overpriced":
return (
f"Edge on {title} invalidates if price moves toward this range, "
f"increasing the probability of landing here."
)
return f"No meaningful edge on {title} — bracket is fairly priced."

def analyze(self, primary_horizon: str = "24h") -> AnalysisResult:
if not self._daily or not self._hourly:
raise ValueError("Both daily and hourly data required for analysis")

edge_24h = self._extract_edge(self._daily, "24h")
edge_1h = self._extract_edge(self._hourly, "1h")

strength = strength_from_horizons(edge_1h.edge_pct, edge_24h.edge_pct)
conflict = signals_conflict(edge_1h.signal, edge_24h.signal)

spread_1h = self._percentile_spread(self._pct_1h)
spread_24h = self._percentile_spread(self._pct_24h)
confidence = self.compute_confidence(spread_1h, spread_24h)

high_uncertainty = any(
s is not None and s > 0.05 for s in (spread_1h, spread_24h)
)
no_trade = conflict or strength == "none" or high_uncertainty

bias_24h = self._directional_bias(self._pct_24h)

explanation = self._build_explanation(edge_1h, edge_24h, confidence)
invalidation = self._build_invalidation(edge_24h, bias_24h)

primary = edge_24h if primary_horizon == "24h" else edge_1h
secondary = edge_1h if primary_horizon == "24h" else edge_24h

return AnalysisResult(
primary=primary,
secondary=secondary,
strength=strength,
confidence_score=confidence,
no_trade=no_trade,
explanation=explanation,
invalidation=invalidation,
)
Loading