chemavxandClaude Fable 5
7ebb87aede
CI/CD / build-and-push (push) Successful in 7s
Bug #5: metrics.record_trade() only delegated to save_trade(), which executor.execute() already calls — every trade was written twice (deduped only by ON CONFLICT DO NOTHING). Remove the redundant call and the now-dead method. RealExecutor.execute() raises NotImplementedError, so real mode is unaffected. Bug #6 (CYCLE SUMMARY): manifold accepted/rejected counters only increment on the active-signal path, so with MANIFOLD_SIGNAL_ENABLED=false they always printed 0/0 — print 'manifold_signal: disabled' instead. family_conflicts_prevented duplicated blocked_by_family (same counter printed twice); removed. gnews_cap was a dead variable with a misleading comment; removed. Bug #7 (/api/summary): total_trades was len() over a LIMIT-500 query — capped once history grows; counts now come from COUNT(*) via compute_metrics_from_db. cash_available was reimplemented in the API; extract cash_available() in paper.py (same formula, unchanged) and feed it from get_open_position_data() — the exact source/helper PaperExecutor.initialize() uses. Test asserts API and executor report identical cash for the same DB state. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
475 lines
19 KiB
Python
475 lines
19 KiB
Python
"""
|
|
Polymarket Trading Bot — Main Entry Point
|
|
# ci-test: 2026-04-16
|
|
"""
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
from datetime import datetime, timezone
|
|
|
|
from bot.data.polymarket import PolymarketClient, Market, market_family_key
|
|
from bot.data.external import ExternalDataClient
|
|
from bot.data.news import NewsClient
|
|
from bot.data.manifold import ManifoldClient
|
|
from bot.strategy.bayesian import (
|
|
BayesianStrategy,
|
|
gnews_priority,
|
|
MAX_NEWS_QUERIES_PER_CYCLE,
|
|
MANIFOLD_SIGNAL_ENABLED,
|
|
)
|
|
from bot.risk.manager import RiskManager
|
|
from bot.executor.paper import PaperExecutor
|
|
from bot.metrics.tracker import MetricsTracker
|
|
from bot.data.db import Database
|
|
from bot.notify.checkpoints import CheckpointMonitor
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
|
)
|
|
log = logging.getLogger("bot.main")
|
|
|
|
PAPER_MODE = os.getenv("PAPER_MODE", "true").lower() == "true"
|
|
PAPER_BANKROLL = float(os.getenv("PAPER_BANKROLL", "10000"))
|
|
|
|
# Check open positions for market resolution every N trading cycles (~N minutes
|
|
# at the 60s cycle cadence). Keeps Gamma API load at ~1 request per open
|
|
# position per 10 minutes.
|
|
RESOLUTION_CHECK_INTERVAL = 10
|
|
|
|
|
|
async def check_resolutions(
|
|
poly: PolymarketClient,
|
|
executor: PaperExecutor,
|
|
db: Database,
|
|
) -> None:
|
|
"""Detect resolved markets and settle their open paper positions.
|
|
|
|
For each open position, asks the Gamma API whether the market resolved.
|
|
On a definitive resolution, PaperExecutor.close_position() settles the
|
|
payout, persists close_reason='resolved' + resolution + close_pnl, and
|
|
sends the Telegram notification.
|
|
"""
|
|
positions = await db.get_open_position_details()
|
|
checked = 0
|
|
resolved = 0
|
|
for pos in positions:
|
|
market_id = str(pos["market_id"])
|
|
try:
|
|
res = await poly.get_market_resolution(market_id)
|
|
except Exception as exc:
|
|
log.warning("Resolution check failed for market %s: %s", market_id, exc)
|
|
continue
|
|
checked += 1
|
|
if res is None or not res.resolved or res.resolution is None:
|
|
continue
|
|
try:
|
|
pnl = await executor.close_position(
|
|
market_id, res.resolution, question=pos.get("question") or "",
|
|
)
|
|
except Exception as exc:
|
|
log.error("Failed to close resolved market %s: %s", market_id, exc)
|
|
continue
|
|
resolved += 1
|
|
log.info(
|
|
"MARKET_RESOLVED market_id=%s resolution=%.1f pnl=%s | %s",
|
|
market_id,
|
|
res.resolution,
|
|
f"{pnl:+.2f}" if pnl is not None else "n/a",
|
|
(pos.get("question") or "")[:60],
|
|
)
|
|
log.info("Resolution check: %d positions checked, %d resolved", checked, resolved)
|
|
|
|
|
|
async def run_trading_loop(
|
|
poly: PolymarketClient,
|
|
external: ExternalDataClient,
|
|
strategy: BayesianStrategy,
|
|
risk: RiskManager,
|
|
executor: PaperExecutor,
|
|
metrics: MetricsTracker,
|
|
db: Database,
|
|
) -> None:
|
|
"""Main trading loop — runs every 60 seconds."""
|
|
log.info("Trading loop started. PAPER_MODE=%s", PAPER_MODE)
|
|
checkpoint_monitor = CheckpointMonitor()
|
|
cycle_count = 0
|
|
|
|
while True:
|
|
try:
|
|
cycle_count += 1
|
|
|
|
# 0. Resolution detector — every RESOLUTION_CHECK_INTERVAL cycles,
|
|
# settle paper positions whose market resolved on Polymarket.
|
|
# Runs before evaluation so freed cash/families are usable this cycle.
|
|
if (
|
|
PAPER_MODE
|
|
and isinstance(executor, PaperExecutor)
|
|
and cycle_count % RESOLUTION_CHECK_INTERVAL == 0
|
|
):
|
|
await check_resolutions(poly, executor, db)
|
|
|
|
# 1. Fetch active markets (90-day window)
|
|
markets = await poly.get_active_markets()
|
|
log.info("Found %d active markets", len(markets))
|
|
|
|
# 2. Get external signals
|
|
ext_data = await external.get_all_signals()
|
|
|
|
# 3. Build occupied_families from the current open portfolio positions.
|
|
# This prevents re-entering a family where we already hold a position.
|
|
# We also pull from DB to survive pod restarts.
|
|
portfolio = executor.get_portfolio()
|
|
occupied_families: set[str] = set()
|
|
for market_id in portfolio.positions:
|
|
mkt = next((m for m in markets if m.id == market_id), None)
|
|
if mkt:
|
|
occupied_families.add(market_family_key(mkt))
|
|
# Also seed from DB in case a family was traded in a prior cycle
|
|
# that isn't reflected in the current markets list
|
|
db_families = await db.get_open_families()
|
|
occupied_families |= db_families
|
|
if occupied_families:
|
|
log.info("Occupied families (from portfolio): %s", sorted(occupied_families))
|
|
|
|
# 4. Sort markets.
|
|
# Politics: sort by gnews_priority DESC (highest-value markets get
|
|
# GNews budget first — Phase 3).
|
|
# Others: sort by end_date ASC (soonest-resolving first).
|
|
def _sort_key(m):
|
|
try:
|
|
dt = datetime.fromisoformat(m.end_date.replace("Z", "+00:00"))
|
|
except Exception:
|
|
dt = datetime(9999, 12, 31, tzinfo=timezone.utc)
|
|
if m.category == "politics":
|
|
priority = gnews_priority(m, strategy._news) if strategy._news else 0.0
|
|
# Bucket 0 = politics, sort by priority DESC (negate for asc sort)
|
|
return (0, -priority, dt)
|
|
return (1, 0.0, dt)
|
|
|
|
markets = sorted(markets, key=_sort_key)
|
|
|
|
for _m in markets:
|
|
log.info(
|
|
" [market] %-55s | cat=%-12s | family=%-28s | ends=%s | yes=%.3f",
|
|
_m.question[:55], _m.category, market_family_key(_m),
|
|
_m.end_date[:10] if _m.end_date else "?", _m.yes_price,
|
|
)
|
|
|
|
# Reset per-cycle GNews counter
|
|
strategy.reset_cycle()
|
|
|
|
# 5. Evaluate each market
|
|
# Fetch markets recently closed for inversion bug — block re-entry for 24h
|
|
inverted_guard: set[str] = await db.get_recently_closed_inverted(hours=24)
|
|
if inverted_guard:
|
|
log.info(
|
|
"Reentry guard active for %d market(s) (inversion, 24h): %s",
|
|
len(inverted_guard), sorted(inverted_guard),
|
|
)
|
|
|
|
reentry_guard_count = 0
|
|
cycle_trades = 0
|
|
for market in markets:
|
|
if market.id in inverted_guard:
|
|
log.info(
|
|
"reentry_guard_triggered market=%s | skipping — closed for inversion within 24h | %s",
|
|
market.id, market.question[:60],
|
|
)
|
|
reentry_guard_count += 1
|
|
continue
|
|
|
|
# evaluate() returns None for all skips — reasons are logged internally
|
|
signal = await strategy.evaluate(market, ext_data, occupied_families)
|
|
if signal is None:
|
|
continue
|
|
|
|
log.info(
|
|
"Signal generated: market=%-50s | edge_gross=%+.3f | edge_net=%+.3f | "
|
|
"regime_min=%.2f | family=%s | conf=%.2f",
|
|
market.question[:50],
|
|
signal.edge_gross,
|
|
signal.edge_net,
|
|
signal.regime_min_edge,
|
|
signal.family_key,
|
|
signal.confidence,
|
|
)
|
|
|
|
# 6. Risk check + position sizing
|
|
order = risk.size_order(signal, portfolio)
|
|
if order is None:
|
|
log.debug("Risk manager rejected order for %s", market.id)
|
|
continue
|
|
|
|
# 7. Execute (paper)
|
|
trade = await executor.execute(order)
|
|
if trade:
|
|
log.info("Trade executed: %s", trade)
|
|
# Block this family for the rest of the cycle (Phase 2)
|
|
occupied_families.add(signal.family_key)
|
|
cycle_trades += 1
|
|
# Mark manifold audit record as used in this trade
|
|
if signal.mfld_audit_id:
|
|
try:
|
|
await db.mark_manifold_audit_used(signal.mfld_audit_id)
|
|
except Exception as exc:
|
|
log.warning("Failed to mark manifold audit used: %s", exc)
|
|
|
|
# 8. [CYCLE SUMMARY] — one block per cycle, stable format for grep/compare
|
|
stats = strategy.get_cycle_stats()
|
|
legacy_incomplete_count = await db.get_legacy_incomplete_count()
|
|
n_total = len(markets)
|
|
n_uncertainty = sum(1 for m in markets if 0.35 <= m.yes_price <= 0.65)
|
|
n_eval = stats["evaluated_count"]
|
|
def _pct(n: int, denom: int) -> str:
|
|
if denom == 0:
|
|
return "0% (0/0)"
|
|
return f"{n * 100 // denom}% ({n}/{denom})"
|
|
|
|
# The accepted/rejected counters only increment on the active-signal
|
|
# path, so with the signal disabled they always print 0/0 — say
|
|
# "disabled" instead of pretending the matcher found nothing.
|
|
if MANIFOLD_SIGNAL_ENABLED:
|
|
manifold_summary = (
|
|
f" manifold_matches_accepted: {stats['manifold_matches_accepted']}\n"
|
|
f" manifold_matches_rejected: {stats['manifold_matches_rejected']}"
|
|
)
|
|
else:
|
|
manifold_summary = " manifold_signal: disabled"
|
|
|
|
log.info(
|
|
"[CYCLE SUMMARY]\n"
|
|
" markets_total: %d\n"
|
|
" markets_uncertainty_zone: %d (prior 0.35-0.65)\n"
|
|
" max_edge_gross: %+.3f\n"
|
|
" max_edge_net: %+.3f\n"
|
|
" pct_edge_gross_gt_002: %s\n"
|
|
" pct_edge_gross_gt_004: %s\n"
|
|
" blocked_by_family: %d\n"
|
|
" blocked_by_prior_extreme: %d\n"
|
|
" blocked_by_edge_net_nonpositive:%d\n"
|
|
" blocked_by_edge_net_below_regime:%d\n"
|
|
" trades_executed: %d\n"
|
|
" gnews_queries_used: %d/%d\n"
|
|
" reentry_guard_blocked: %d\n"
|
|
" legacy_incomplete_seen: %d\n"
|
|
"%s",
|
|
n_total,
|
|
n_uncertainty,
|
|
stats["max_edge_gross"],
|
|
stats["max_edge_net"],
|
|
_pct(stats["gross_gt_002"], n_total),
|
|
_pct(stats["gross_gt_004"], n_total),
|
|
stats["skip_family"],
|
|
stats["skip_prior_extreme"],
|
|
stats["skip_edge_net_nonpositive"],
|
|
stats["skip_edge_net_below_regime"],
|
|
cycle_trades,
|
|
stats["gnews_queries_used"], MAX_NEWS_QUERIES_PER_CYCLE,
|
|
reentry_guard_count,
|
|
legacy_incomplete_count,
|
|
manifold_summary,
|
|
)
|
|
|
|
# 9. Update daily metrics
|
|
await metrics.update_daily_summary()
|
|
|
|
# 10. Checkpoint alerts — one-shot / rate-limited Telegram notifications
|
|
current_portfolio = executor.get_portfolio()
|
|
try:
|
|
await checkpoint_monitor.check_all(
|
|
db,
|
|
exposure_pct=current_portfolio.exposure_pct,
|
|
exposure_cap_pct=risk.max_exposure_pct,
|
|
)
|
|
except Exception as exc:
|
|
log.warning("checkpoint_monitor.check_all failed: %s", exc)
|
|
|
|
except Exception as e:
|
|
log.error("Error in trading loop: %s", e, exc_info=True)
|
|
|
|
await asyncio.sleep(60)
|
|
|
|
|
|
async def run_legacy_scan(
|
|
db: Database,
|
|
markets: list,
|
|
executor: PaperExecutor,
|
|
paper_mode: bool,
|
|
) -> None:
|
|
"""
|
|
One-time startup scan: re-key all open DB positions with the current
|
|
market_family_key() logic, detect family conflicts, and report
|
|
KEEP / REVIEW / CLOSE_RECOMMENDED per position.
|
|
|
|
Manifold is intentionally not consulted here: with
|
|
MANIFOLD_SIGNAL_ENABLED=false it is observational-only and must not
|
|
drive position closures.
|
|
|
|
In paper_mode: auto-closes all CLOSE_RECOMMENDED positions after logging.
|
|
"""
|
|
positions = await db.get_open_position_details()
|
|
if not positions:
|
|
log.info("Legacy scan: no open positions — skipping.")
|
|
return
|
|
|
|
market_by_id: dict = {str(m.id): m for m in markets}
|
|
|
|
# Step 1: enrich each position with the re-computed family key
|
|
enriched: list[dict] = []
|
|
for pos in positions:
|
|
mid = str(pos["market_id"])
|
|
live_mkt = market_by_id.get(mid)
|
|
old_fk = pos.get("family_key") or ""
|
|
if live_mkt:
|
|
new_fk = market_family_key(live_mkt)
|
|
else:
|
|
# Market not in active list — compute from stored question alone
|
|
_dummy = Market(
|
|
id=mid, condition_id="", question=pos["question"],
|
|
yes_token_id="", no_token_id="",
|
|
yes_price=0.5, no_price=0.5,
|
|
volume_24h=0, end_date="", active=False, category="",
|
|
)
|
|
computed = market_family_key(_dummy)
|
|
# Reject degenerate fallbacks that start with "-" (missing category + end_date)
|
|
new_fk = computed if not computed.startswith("-") else (old_fk or "unknown")
|
|
|
|
is_legacy_incomplete = (pos.get("edge_net") is None) and (not live_mkt)
|
|
enriched.append({
|
|
**dict(pos),
|
|
"market_id": mid,
|
|
"live_market": live_mkt,
|
|
"family_key_old": old_fk,
|
|
"family_key_new": new_fk,
|
|
"fk_changed": new_fk != old_fk,
|
|
"recommendation": "legacy_incomplete" if is_legacy_incomplete else "OK",
|
|
"rec_reason": "edge_net and live market unavailable" if is_legacy_incomplete else "no family conflict",
|
|
})
|
|
|
|
# Step 2: group by new family key — identify conflicting siblings
|
|
family_groups: dict[str, list[dict]] = {}
|
|
for p in enriched:
|
|
family_groups.setdefault(p["family_key_new"], []).append(p)
|
|
|
|
for p in enriched:
|
|
group = family_groups[p["family_key_new"]]
|
|
if len(group) > 1:
|
|
best = max(group, key=lambda x: (x.get("edge_net") or 0.0))
|
|
if p["market_id"] == best["market_id"]:
|
|
p["recommendation"] = "KEEP"
|
|
p["rec_reason"] = (
|
|
f"highest edge_net={p.get('edge_net') or 0.0:.3f} in family"
|
|
)
|
|
else:
|
|
p["recommendation"] = "CLOSE_RECOMMENDED"
|
|
p["rec_reason"] = (
|
|
f"family conflict: sibling {best['market_id']} "
|
|
f"has edge_net={best.get('edge_net') or 0.0:.3f}"
|
|
)
|
|
elif p["fk_changed"]:
|
|
p["recommendation"] = "REVIEW"
|
|
p["rec_reason"] = "family key changed but no sibling conflict"
|
|
|
|
# Step 2.5: persist corrected family keys in DB for changed positions
|
|
for p in enriched:
|
|
if p["fk_changed"] and p["family_key_new"] not in ("unknown", ""):
|
|
await db.update_family_key(p["market_id"], p["family_key_new"])
|
|
log.info(
|
|
"family_key updated in DB: market=%s | %s → %s",
|
|
p["market_id"], p["family_key_old"] or "none", p["family_key_new"],
|
|
)
|
|
|
|
# Step 3: log the full scan report (before any closures)
|
|
n_close = sum(1 for p in enriched if p["recommendation"] == "CLOSE_RECOMMENDED")
|
|
n_keep = sum(1 for p in enriched if p["recommendation"] == "KEEP")
|
|
n_ok = sum(1 for p in enriched if p["recommendation"] == "OK")
|
|
n_review = sum(1 for p in enriched if p["recommendation"] == "REVIEW")
|
|
n_legacy = sum(1 for p in enriched if p["recommendation"] == "legacy_incomplete")
|
|
|
|
log.warning(
|
|
"━" * 70 + "\nLEGACY SCAN — %d position(s): OK=%d KEEP=%d REVIEW=%d CLOSE_RECOMMENDED=%d LEGACY_INCOMPLETE=%d",
|
|
len(enriched), n_ok, n_keep, n_review, n_close, n_legacy,
|
|
)
|
|
for p in enriched:
|
|
log.warning(
|
|
" [%-18s] market=%-8s | dir=%-8s | edge_net=%+.3f\n"
|
|
" stored_family: %s\n"
|
|
" new_family: %s%s\n"
|
|
" reason: %s",
|
|
p["recommendation"],
|
|
p["market_id"], p["direction"],
|
|
p.get("edge_net") or 0.0,
|
|
p["family_key_old"] or "none",
|
|
p["family_key_new"],
|
|
" [CHANGED]" if p["fk_changed"] else "",
|
|
p["rec_reason"],
|
|
)
|
|
log.warning("━" * 70)
|
|
|
|
# Step 4: auto-close in paper mode
|
|
if paper_mode and n_close > 0 and isinstance(executor, PaperExecutor):
|
|
log.warning("PAPER MODE: auto-closing %d CLOSE_RECOMMENDED position(s)...", n_close)
|
|
for p in enriched:
|
|
if p["recommendation"] == "CLOSE_RECOMMENDED":
|
|
recovered = await executor.close_legacy_position(p["market_id"], p["rec_reason"], question=p["question"])
|
|
log.warning(
|
|
" AUTO_CLOSED market=%s | $%.2f returned to cash | %s",
|
|
p["market_id"], recovered, p["question"][:60],
|
|
)
|
|
log.warning("Legacy scan closures complete.")
|
|
elif n_close > 0:
|
|
log.warning("REAL MODE: %d position(s) marked CLOSE_RECOMMENDED — close manually.", n_close)
|
|
|
|
|
|
async def main() -> None:
|
|
if PAPER_MODE:
|
|
log.info("=" * 60)
|
|
log.info(" PAPER TRADING MODE — No real money at risk")
|
|
log.info(" Bankroll: $%.2f simulated", PAPER_BANKROLL)
|
|
log.info("=" * 60)
|
|
else:
|
|
log.warning("REAL TRADING MODE ACTIVE — Real money at risk!")
|
|
|
|
db = Database()
|
|
await db.connect()
|
|
await db.run_migrations()
|
|
await db.backfill_feature_columns()
|
|
|
|
poly = PolymarketClient()
|
|
external = ExternalDataClient()
|
|
news = NewsClient()
|
|
manifold = ManifoldClient()
|
|
strategy = BayesianStrategy(news=news, manifold=manifold, db=db)
|
|
risk = RiskManager(max_position_pct=0.05, max_exposure_pct=0.30)
|
|
executor = PaperExecutor(db=db, bankroll=PAPER_BANKROLL) if PAPER_MODE else None
|
|
metrics = MetricsTracker(db=db)
|
|
|
|
if executor is None:
|
|
from bot.executor.real import RealExecutor # noqa
|
|
executor = RealExecutor(db=db)
|
|
|
|
if PAPER_MODE:
|
|
await executor.initialize()
|
|
|
|
# Legacy scan: re-key all open positions, detect contradictions, auto-close
|
|
# CLOSE_RECOMMENDED in paper mode. Runs once at startup using a fresh
|
|
# market snapshot; the trading loop will re-fetch on its own first cycle.
|
|
try:
|
|
scan_markets = await poly.get_active_markets()
|
|
except Exception as e:
|
|
log.warning("Could not fetch markets for legacy scan: %s — scan skipped", e)
|
|
scan_markets = []
|
|
await run_legacy_scan(db, scan_markets, executor, PAPER_MODE)
|
|
|
|
try:
|
|
await run_trading_loop(poly, external, strategy, risk, executor, metrics, db)
|
|
finally:
|
|
await db.disconnect()
|
|
await news.close()
|
|
await manifold.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|