gavindiaz/my-polymarket-copytrader
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

全面优化实盘对接没有验证

Browse Source
This commit is contained in:
gavindiaz
2026-07-14 07:29:03 +08:00
parent a1963e58ed
commit b3f770c55f
391 changed files with 114674 additions and 528 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/config/settings.py
+48 -4
View File
@@ -26,19 +26,34 @@ class Settings(BaseSettings):
# Wallet pool
wallet_pool_size: int = Field(default=100, alias="WALLET_POOL_SIZE")
wallet_pnl_min_usd: float = Field(default=5000.0, alias="WALLET_PNL_MIN_USD")
wallet_pnl_min_usd: float = Field(default=5000, alias="WALLET_PNL_MIN_USD")
wallet_pnl_window_days: int = Field(default=90, alias="WALLET_PNL_WINDOW_DAYS")
wallet_min_trades: int = Field(default=30, alias="WALLET_MIN_TRADES")
wallet_min_categories: int = Field(default=3, alias="WALLET_MIN_CATEGORIES")
wallet_refresh_hours: int = Field(default=24, alias="WALLET_REFRESH_HOURS")
# Max pages for /closed-positions pagination (each page = 50 records).
# Pool building uses this default; bayesian updater uses a higher value
# since it only processes ~100 pool wallets and needs PnL accuracy.
# 20 pages = 1000 records cap. Only ~0.1% of wallets exceed this.
closed_positions_max_pages: int = Field(default=20, alias="CLOSED_POSITIONS_MAX_PAGES")
# Discount applied to unrealized cashPnl when scoring wallet health.
# Polymarket mid-prices often overstate actual fillable value due to
# thin liquidity. 0.5 = only count half of unrealized PnL.
cash_pnl_discount: float = Field(default=0.5, alias="CASH_PNL_DISCOUNT")
# Exclude wallets that hold both Yes AND No on the same market — these
# are typically market makers / hedgers whose trades are not directional.
exclude_market_makers: bool = Field(default=True, alias="EXCLUDE_MARKET_MAKERS")
# Bayesian credibility priors
bayesian_prior_skill: float = Field(default=0.5, alias="BAYESIAN_PRIOR_SKILL")
bayesian_decay_days: int = Field(default=14, alias="BAYESIAN_DECAY_DAYS")
bayesian_step: float = Field(default=0.05, alias="BAYESIAN_STEP")
# User stream polling (Phase 1)
user_poll_interval_seconds: int = Field(default=30, alias="USER_POLL_INTERVAL_SECONDS")
stream_warmup_seconds: int = Field(default=30, alias="STREAM_WARMUP_SECONDS")
stream_max_trades_per_wallet: int = Field(default=20, alias="STREAM_MAX_TRADES_PER_WALLET")
stream_verbose_logging: bool = Field(default=False, alias="STREAM_VERBOSE_LOGGING")
# Aggregator (Phase 2)
consensus_window_seconds: int = Field(default=600, alias="CONSENSUS_WINDOW_SECONDS")
@@ -46,19 +61,48 @@ class Settings(BaseSettings):
consensus_strength_threshold: float = Field(default=0.4, alias="CONSENSUS_STRENGTH_THRESHOLD")
wallet_debounce_seconds: int = Field(default=600, alias="WALLET_DEBOUNCE_SECONDS")
min_credibility: float = Field(default=0.3, alias="MIN_CREDIBILITY")
# Only emit BUY signals. SELL on Polymarket = selling tokens you already
# hold; without position tracking we cannot follow SELLs. SELL trades
# are still logged as "reverse indicator" but do NOT trigger signals.
buy_only_signals: bool = Field(default=True, alias="BUY_ONLY_SIGNALS")
# Same-market signal cooldown: once a signal is emitted for condition_id,
# suppress new signals on that market for N seconds.
signal_cooldown_seconds: int = Field(default=86400, alias="SIGNAL_COOLDOWN_SECONDS")
# Skip markets that resolve within N hours — thin liquidity + high
# resolution risk makes them dangerous to follow.
min_hours_to_resolution: int = Field(default=24, alias="MIN_HOURS_TO_RESOLUTION")
# Kelly fallback: when historical hit-rate data is insufficient (<N
# resolved signals), use a conservative fixed position fraction instead
# of the Kelly formula.
kelly_min_samples: int = Field(default=50, alias="KELLY_MIN_SAMPLES")
kelly_fallback_fraction: float = Field(default=0.01, alias="KELLY_FALLBACK_FRACTION")
# Risk management
max_daily_loss_usd: float = Field(default=500.0, alias="MAX_DAILY_LOSS_USD")
max_consecutive_losses: int = Field(default=5, alias="MAX_CONSECUTIVE_LOSSES")
risk_pause_minutes: int = Field(default=60, alias="RISK_PAUSE_MINUTES")
max_open_positions: int = Field(default=10, alias="MAX_OPEN_POSITIONS")
# Credibility update loop
credibility_update_minutes: int = Field(default=60, alias="CREDIBILITY_UPDATE_MINUTES")
# Outcome resolution (signal backfill)
outcome_resolve_minutes: int = Field(default=15, alias="OUTCOME_RESOLVE_MINUTES")
outcome_resolve_batch_size: int = Field(default=50, alias="OUTCOME_RESOLVE_BATCH_SIZE")
outcome_resolve_request_timeout: float = Field(default=5.0, alias="OUTCOME_RESOLVE_TIMEOUT")
# Debug logging (extra verbose beyond LOG_LEVEL)
debug_log_api_payloads: bool = Field(default=False, alias="DEBUG_LOG_API_PAYLOADS")
# Pool builder concurrency
wallet_pool_concurrency: int = Field(default=15, alias="WALLET_POOL_CONCURRENCY")
wallet_pool_concurrency: int = Field(default=30, alias="WALLET_POOL_CONCURRENCY")
wallet_pool_request_timeout: float = Field(default=5.0, alias="WALLET_POOL_REQUEST_TIMEOUT")
wallet_pool_progress_every: int = Field(default=50, alias="WALLET_POOL_PROGRESS_EVERY")
wallet_pool_progress_every: int = Field(default=500, alias="WALLET_POOL_PROGRESS_EVERY")
wallet_pool_backoff_emails: int = Field(default=30, alias="WALLET_POOL_BACKOFF_EMA") # consecutive empty responses → sleep
# Bayesian credibility updater concurrency
bayes_concurrency: int = Field(default=10, alias="BAYES_CONCURRENCY")
# Telegram notifications
telegram_enabled: bool = Field(default=False, alias="TELEGRAM_ENABLED")
telegram_bot_token: str = Field(default="", alias="TELEGRAM_BOT_TOKEN")
@@ -86,4 +130,4 @@ class Settings(BaseSettings):
@lru_cache
def get_settings() -> Settings:
load_dotenv()
return Settings()
return Settings()
src/dashboard.py
+271 -22
View File
@@ -5,6 +5,14 @@ Endpoints:
/api/wallets — top-priority wallet pool
/api/signals/recent — recent consensus signals
/api/health — liveness check
/api/outcomes — overall signal outcome stats (hit rate, PnL)
/api/stats/by-strength — hit rate bucketed by aggregated_strength
/api/stats/by-wallet-count — hit rate bucketed by n_contributors
/api/stats/by-wallet — per-wallet hit rate (reverse-validate credibility)
/api/stats/by-category — per-category hit rate
/api/stats/by-hour — per-hour-of-day hit rate
/api/credibility-history?address=... — credibility evolution curve
/analytics — strategy-optimization dashboard (HTML)
"""
import logging
from typing import Optional
@@ -19,28 +27,31 @@ logger = logging.getLogger(__name__)
app = FastAPI(title="Polymarket Copy Trader Dashboard")
_db_instance: CopyTraderDatabase | None = None
def _get_db() -> CopyTraderDatabase:
settings = get_settings()
return CopyTraderDatabase(settings.db_path)
global _db_instance
if _db_instance is None:
settings = get_settings()
_db_instance = CopyTraderDatabase(settings.db_path)
return _db_instance
@app.get("/api/stats")
def api_stats():
db = _get_db()
return db.get_stats()
return _get_db().get_stats()
@app.get("/api/wallets")
def api_wallets(limit: int = Query(50, ge=1, le=200)):
db = _get_db()
wallets = db.get_all_wallet_targets()
wallets = _get_db().get_all_wallet_targets()
return wallets[:limit]
@app.get("/api/signals/recent")
def api_signals(limit: int = Query(20, ge=1, le=200)):
db = _get_db()
return db.get_recent_signals(limit=limit)
return _get_db().get_recent_signals(limit=limit)
@app.get("/api/health")
@@ -48,12 +59,53 @@ def api_health():
return {"status": "ok"}
# ----- Strategy optimization endpoints -----
@app.get("/api/outcomes")
def api_outcomes():
"""Overall hit rate + PnL distribution."""
return _get_db().get_outcome_stats()
@app.get("/api/stats/by-strength")
def api_stats_by_strength():
return _get_db().get_signal_stats_by_strength()
@app.get("/api/stats/by-wallet-count")
def api_stats_by_wallet_count():
return _get_db().get_signal_stats_by_wallet_count()
@app.get("/api/stats/by-wallet")
def api_stats_by_wallet(limit: int = Query(30, ge=1, le=200)):
return _get_db().get_signal_stats_by_wallet(limit=limit)
@app.get("/api/stats/by-category")
def api_stats_by_category(limit: int = Query(30, ge=1, le=100)):
return _get_db().get_signal_stats_by_category(limit=limit)
@app.get("/api/stats/by-hour")
def api_stats_by_hour():
return _get_db().get_signal_stats_by_hour()
@app.get("/api/credibility-history")
def api_credibility_history(
address: str = Query(..., min_length=42, max_length=42),
limit: int = Query(200, ge=1, le=2000),
):
return _get_db().get_credibility_history(address, limit=limit)
HTML_TEMPLATE = """<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Polymarket Copy Trader Dashboard</title>
<title>Polymarket 跟单交易看板</title>
<style>
* { margin: 0; padding: 0; box-sizing: border-box; }
body { font-family: -apple-system, sans-serif; background: #0f1117; color: #e0e0e0; padding: 20px; }
@@ -74,27 +126,29 @@ tr:hover { background: #1a1d28; }
.b-low { background: #88888833; color: #aaa; }
.mono { font-family: monospace; font-size: 0.85em; color: #888; }
#loading { color: #666; text-align: center; padding: 40px; }
a { color: #4fc3f7; text-decoration: none; }
a:hover { text-decoration: underline; }
</style>
</head>
<body>
<h1>Polymarket Copy Trader</h1>
<p class="subtitle" id="update-info">Live signal & wallet dashboard · auto-refresh 30s</p>
<p class="subtitle" id="update-info">实时信号与钱包看板 · 每30秒自动刷新</p>
<div id="loading">Loading...</div>
<div id="loading">加载中...</div>
<div id="content" style="display:none">
<div class="stats" id="stats"></div>
<h2>Top Wallets</h2>
<h2>优质钱包榜</h2>
<table id="wallet-table">
<thead><tr><th>Rank</th><th>Address</th><th>PnL Total</th><th>Trades</th><th>Categories</th><th>Health</th><th>Credibility</th></tr></thead>
<thead><tr><th>排名</th><th>地址</th><th>总盈亏</th><th>交易数</th><th>品类数</th><th>健康分</th><th>信誉分</th></tr></thead>
<tbody></tbody>
</table>
<h2>Recent Signals</h2>
<h2>最近信号</h2>
<table id="signal-table">
<thead><tr><th>Time</th><th>Market</th><th>Direction</th><th>Entry</th><th>Strength</th><th>Wallets</th><th>Kelly Size</th></tr></thead>
<thead><tr><th>时间</th><th>市场</th><th>方向</th><th>入场价</th><th>强度</th><th>钱包数</th><th>建议仓位</th></tr></thead>
<tbody></tbody>
</table>
@@ -110,9 +164,9 @@ async function load() {
]);
document.getElementById('stats').innerHTML = [
['Wallets', statsR.wallet_count],
['Total Signals', statsR.total_signals],
['Executed', statsR.executed_signals]
['钱包数', statsR.wallet_count],
['信号总数', statsR.total_signals],
['已执行', statsR.executed_signals]
].map(([l,v]) => `<div class="card"><div class="v">${v}</div><div class="l">${l}</div></div>`).join('');
document.querySelector('#wallet-table tbody').innerHTML = walletsR.map((w,i) =>
@@ -131,11 +185,11 @@ async function load() {
const cls = s.aggregated_strength > 0.7 ? 'b-high' : s.aggregated_strength > 0.4 ? 'b-med' : 'b-low';
return `<tr>
<td>${(s.generated_at||'').slice(0,16)}</td>
<td>${(s.market_question||'').slice(0,60)}</td>
<td><a href="https://polymarket.com/market/${s.market_slug||s.condition_id||''}" target="_blank" title="Open on Polymarket">${(s.market_question||'Unknown Market').slice(0,60)}</a></td>
<td>${s.side} ${s.outcome}</td>
<td>${Number(s.entry_price||0).toFixed(4)}</td>
<td><span class="badge ${cls}">${(s.aggregated_strength*100).toFixed(0)}%</span></td>
<td>${s.n_contributors||'?'}</td>
<td>${s.n_contributors ?? '?'}</td>
<td>$${Number(s.suggested_size_usd||0).toFixed(0)}</td>
</tr>`;
}).join('');
@@ -143,9 +197,9 @@ async function load() {
document.getElementById('loading').style.display = 'none';
document.getElementById('content').style.display = 'block';
document.getElementById('update-info').textContent =
`Last refresh: ${new Date().toLocaleTimeString()} · auto-refresh 30s`;
`上次刷新：${new Date().toLocaleTimeString()} · 每30秒自动刷新`;
} catch(e) {
document.getElementById('loading').textContent = 'Error: ' + e.message;
document.getElementById('loading').textContent = '错误：' + e.message;
}
}
load();
@@ -159,3 +213,198 @@ setInterval(load, 30000);
@app.get("/", response_class=HTMLResponse)
def dashboard_page():
return HTML_TEMPLATE
ANALYTICS_TEMPLATE = """<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>策略分析 · Polymarket Copy Trader</title>
<style>
* { margin: 0; padding: 0; box-sizing: border-box; }
body { font-family: -apple-system, sans-serif; background: #0f1117; color: #e0e0e0; padding: 20px; }
h1 { color: #fff; margin-bottom: 8px; }
h2 { color: #a0a8c0; margin: 32px 0 12px; font-size: 1.2em; }
.subtitle { color: #666; margin-bottom: 24px; }
a { color: #4fc3f7; text-decoration: none; }
a:hover { text-decoration: underline; }
.stats { display: grid; grid-template-columns: repeat(auto-fit, minmax(160px, 1fr)); gap: 12px; margin-bottom: 24px; }
.card { background: #1a1d28; border-radius: 8px; padding: 16px; text-align: center; }
.card .v { font-size: 1.8em; font-weight: bold; color: #4fc3f7; }
.card .v.pos { color: #66bb6a; }
.card .v.neg { color: #ef5350; }
.card .l { color: #888; font-size: 0.85em; margin-top: 4px; }
table { width: 100%; border-collapse: collapse; margin-bottom: 24px; }
th { background: #1a1d28; color: #a0a8c0; text-align: left; padding: 10px 12px; }
td { padding: 10px 12px; border-bottom: 1px solid #222; font-size: 0.9em; }
tr:hover { background: #1a1d28; }
.bar-row { display: flex; align-items: center; gap: 8px; margin-bottom: 6px; }
.bar-row .label { width: 80px; text-align: right; color: #888; font-size: 0.85em; }
.bar-row .bar-bg { flex: 1; background: #222; border-radius: 4px; height: 22px; position: relative; }
.bar-row .bar-fill { height: 100%; border-radius: 4px; background: #4fc3f7; }
.bar-row .bar-fill.pos { background: #66bb6a; }
.bar-row .bar-fill.neg { background: #ef5350; }
.bar-row .bar-text { position: absolute; left: 8px; top: 2px; font-size: 0.8em; color: #fff; }
.mono { font-family: monospace; font-size: 0.85em; color: #888; }
#loading { color: #666; text-align: center; padding: 40px; }
.nav { margin-bottom: 24px; }
.nav a { margin-right: 16px; }
</style>
</head>
<body>
<h1>策略分析</h1>
<p class="subtitle">信号命中率与 PnL 聚合 · 用于调参</p>
<div class="nav">
<a href="/">← 返回主看板</a>
</div>
<div id="loading">加载中...</div>
<div id="content" style="display:none">
<h2>总体表现</h2>
<div class="stats" id="outcome-stats"></div>
<h2>强度分桶命中率</h2>
<p class="subtitle">用于校准 CONSENSUS_STRENGTH_THRESHOLD。理想情况下强度越高命中率越高。</p>
<div id="by-strength"></div>
<h2>共识钱包数分桶</h2>
<p class="subtitle">用于校准 CONSENSUS_MIN_WALLETS。理想情况下钱包数越多命中率越高。</p>
<div id="by-wallet-count"></div>
<h2>各时段命中率</h2>
<p class="subtitle">识别活跃时段与高命中时段。</p>
<div id="by-hour"></div>
<h2>品类表现</h2>
<p class="subtitle">不同品类的命中率差异(基于 market_slug 前缀)。</p>
<table id="by-category-table">
<thead><tr><th>品类</th><th>信号数</th><th>胜</th><th>命中率</th><th>总 PnL</th></tr></thead>
<tbody></tbody>
</table>
<h2>钱包排行榜</h2>
<p class="subtitle">反向验证信誉分公式 — 高信誉钱包应该排在这里的前列。</p>
<table id="by-wallet-table">
<thead><tr><th>排名</th><th>钱包</th><th>信号数</th><th>胜</th><th>命中率</th><th>总 PnL</th></tr></thead>
<tbody></tbody>
</table>
</div>
<script>
function fmt(n, d=0) { return Number(n||0).toLocaleString('en-US', {maximumFractionDigits:d}); }
function pct(n, d=1) { return (Number(n||0)*100).toFixed(d) + '%'; }
function pnl(n) {
const v = Number(n||0);
const sign = v >= 0 ? '+' : '';
return `${sign}$${fmt(v, 2)}`;
}
function renderBar(label, value, max, textClass='') {
const width = max > 0 ? (Math.abs(value) / max * 100) : 0;
const cls = value >= 0 ? 'pos' : 'neg';
return `<div class="bar-row">
<span class="label">${label}</span>
<div class="bar-bg"><div class="bar-fill ${cls}" style="width:${width}%"></div>
<span class="bar-text">${textClass || ''}</span>
</div>
</div>`;
}
async function load() {
try {
const [outcomes, strength, walletCount, hour, cat, wallet] = await Promise.all([
fetch('/api/outcomes').then(r => r.json()),
fetch('/api/stats/by-strength').then(r => r.json()),
fetch('/api/stats/by-wallet-count').then(r => r.json()),
fetch('/api/stats/by-hour').then(r => r.json()),
fetch('/api/stats/by-category').then(r => r.json()),
fetch('/api/stats/by-wallet?limit=30').then(r => r.json()),
]);
// Overall stats
const o = outcomes;
const hitRateCls = o.hit_rate >= 0.55 ? 'pos' : (o.hit_rate < 0.45 ? 'neg' : '');
const pnlCls = o.total_pnl_usd >= 0 ? 'pos' : 'neg';
document.getElementById('outcome-stats').innerHTML = [
['信号总数', o.total_signals, ''],
['已 resolve', o.wins + o.losses, ''],
['胜', o.wins, 'pos'],
['负', o.losses, 'neg'],
['命中率', pct(o.hit_rate), hitRateCls],
['总 PnL', pnl(o.total_pnl_usd), pnlCls],
['平均 PnL', pnl(o.avg_pnl_usd), pnlCls],
['待 resolve', o.pending, ''],
].map(([l,v,cls]) => `<div class="card"><div class="v ${cls}">${v}</div><div class="l">${l}</div></div>`).join('');
// By strength
const maxStrengthPnl = Math.max(...strength.map(s => Math.abs(s.pnl||0)), 1);
document.getElementById('by-strength').innerHTML = strength.map(s => {
const hr = s.n > 0 ? s.wins / s.n : 0;
return renderBar(s.bucket, s.pnl, maxStrengthPnl,
`${s.n} 信号 · 命中 ${pct(hr)} · PnL ${pnl(s.pnl)}`);
}).join('');
// By wallet count
const maxWcPnl = Math.max(...walletCount.map(s => Math.abs(s.pnl||0)), 1);
document.getElementById('by-wallet-count').innerHTML = walletCount.map(s => {
const hr = s.n > 0 ? s.wins / s.n : 0;
return renderBar(`${s.bucket}`, s.pnl, maxWcPnl,
`${s.n} 信号 · 命中 ${pct(hr)} · PnL ${pnl(s.pnl)}`);
}).join('');
// By hour
const maxHourPnl = Math.max(...hour.map(s => Math.abs(s.pnl||0)), 1);
document.getElementById('by-hour').innerHTML = hour.map(s => {
const hr = s.n > 0 ? s.wins / s.n : 0;
return renderBar(`${String(s.hour).padStart(2,'0')}:00`, s.pnl, maxHourPnl,
`${s.n} 信号 · 命中 ${pct(hr)} · PnL ${pnl(s.pnl)}`);
}).join('');
// By category
document.querySelector('#by-category-table tbody').innerHTML = cat.map(c => {
const hr = c.n > 0 ? c.wins / c.n : 0;
return `<tr>
<td>${c.category}</td>
<td>${c.n}</td>
<td>${c.wins}</td>
<td>${pct(hr)}</td>
<td>${pnl(c.pnl)}</td>
</tr>`;
}).join('');
// By wallet
document.querySelector('#by-wallet-table tbody').innerHTML = wallet.map((w, i) => {
const hr = w.n_signals > 0 ? w.wins / w.n_signals : 0;
return `<tr>
<td>${i+1}</td>
<td class="mono">${(w.address||'').slice(0,10)}...${(w.address||'').slice(-4)}</td>
<td>${w.n_signals}</td>
<td>${w.wins}</td>
<td>${pct(hr)}</td>
<td>${pnl(w.pnl)}</td>
</tr>`;
}).join('');
document.getElementById('loading').style.display = 'none';
document.getElementById('content').style.display = 'block';
} catch(e) {
document.getElementById('loading').textContent = '错误：' + e.message;
}
}
load();
</script>
</body>
</html>"""
@app.get("/analytics", response_class=HTMLResponse)
def analytics_page():
"""Strategy-optimization dashboard: hit rate by strength / wallet count /
hour / category / wallet, plus overall PnL stats."""
return ANALYTICS_TEMPLATE
src/db/database.py
+565 -161
View File
@@ -1,225 +1,455 @@
"""SQLite storage for wallet pool, signals, and trade executions."""
"""SQLite storage for wallet pool, signals, and trade executions.
Concurrency model
-----------------
This DB is shared across multiple asyncio coroutines (pool_refresh,
trade_stream, credibility_updater) that reach it via ``asyncio.to_thread``.
Each call therefore runs in a thread-pool worker, which means the underlying
sqlite3 connection can be touched from several OS threads concurrently.
To make this safe and fast we:
* Open ONE long-lived connection with ``check_same_thread=False``.
* Guard every operation with a single ``threading.RLock``. This serializes
Python-level access, so within this process there is never more than one
writer. (SQLite/WAL still only allows one writer at a time anyway.)
* Enable WAL + ``synchronous=NORMAL`` + ``busy_timeout=10000ms``. The busy
timeout covers cross-process contention (e.g. the dashboard process
writing to the same DB file) by waiting instead of raising
``database is locked`` immediately.
The previous design opened a fresh connection per call; under concurrent
writers that triggered ``database is locked`` because the default
``busy_timeout`` is 0.
"""
import json
import logging
import sqlite3
from datetime import datetime
import threading
from datetime import datetime, timezone
from pathlib import Path
from typing import List, Optional
logger = logging.getLogger(__name__)
def _now_iso_utc() -> str:
"""UTC ISO-8601 timestamp with offset, e.g. '2026-07-14T07:30:00+00:00'.
Used for all DB datetime columns so that string sorting matches
chronological order regardless of the host's local timezone.
"""
return datetime.now(timezone.utc).isoformat()
class CopyTraderDatabase:
"""SQLite storage for the copy-trader."""
def __init__(self, db_path: str = "data/copytrader.db"):
self.db_path = Path(db_path)
self.db_path.parent.mkdir(parents=True, exist_ok=True)
# RLock so nested helper calls within the same thread don't deadlock.
self._lock = threading.RLock()
# Single long-lived connection; check_same_thread=False because
# asyncio.to_thread dispatches us to arbitrary worker threads.
self._conn = sqlite3.connect(
str(self.db_path),
check_same_thread=False,
timeout=10.0, # also sets busy_timeout internally
isolation_level="DEFERRED",
)
self._conn.row_factory = sqlite3.Row
# WAL: multi-reader + single-writer, no reader/writer blocking.
self._conn.execute("PRAGMA journal_mode=WAL")
# NORMAL is safe under WAL and far faster than FULL.
self._conn.execute("PRAGMA synchronous=NORMAL")
# Cross-process writer contention: wait up to 10s instead of failing.
self._conn.execute("PRAGMA busy_timeout=10000")
self._conn.execute("PRAGMA foreign_keys=ON")
self._init_db()
def _get_conn(self) -> sqlite3.Connection:
conn = sqlite3.connect(str(self.db_path))
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL")
return conn
def close(self) -> None:
with self._lock:
try:
self._conn.commit()
except Exception:
pass
self._conn.close()
# ----- Schema -----
def _init_db(self):
with self._get_conn() as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS wallet_targets (
address TEXT PRIMARY KEY,
source TEXT NOT NULL,
pnl_30d_usd REAL DEFAULT 0,
pnl_total_usd REAL DEFAULT 0,
trades_count INTEGER DEFAULT 0,
categories_count INTEGER DEFAULT 0,
health_score REAL DEFAULT 0,
credibility REAL DEFAULT 0.5,
last_seen_at TEXT,
added_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS copy_signals (
id INTEGER PRIMARY KEY AUTOINCREMENT,
condition_id TEXT NOT NULL,
market_question TEXT,
side TEXT NOT NULL,
outcome TEXT NOT NULL,
entry_price REAL NOT NULL,
aggregated_strength REAL NOT NULL,
source_wallets_json TEXT NOT NULL,
kelly_fraction REAL NOT NULL,
suggested_size_usd REAL NOT NULL,
generated_at TEXT NOT NULL,
executed INTEGER DEFAULT 0,
exit_price REAL,
pnl_usd REAL,
resolved_at TEXT
)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_copy_signals_condition
ON copy_signals(condition_id)
""")
conn.execute("""
CREATE INDEX IF NOT EXISTS idx_copy_signals_generated_at
ON copy_signals(generated_at DESC)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS trade_executions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
signal_id INTEGER,
condition_id TEXT NOT NULL,
side TEXT NOT NULL,
outcome TEXT NOT NULL,
size_usd REAL NOT NULL,
price REAL NOT NULL,
order_id TEXT,
status TEXT NOT NULL,
error TEXT,
executed_at TEXT NOT NULL,
FOREIGN KEY(signal_id) REFERENCES copy_signals(id)
)
""")
with self._lock:
with self._conn: # transaction: commit/rollback on exit
self._conn.execute("""
CREATE TABLE IF NOT EXISTS wallet_targets (
address TEXT PRIMARY KEY,
source TEXT NOT NULL,
pnl_30d_usd REAL DEFAULT 0,
pnl_total_usd REAL DEFAULT 0,
trades_count INTEGER DEFAULT 0,
categories_count INTEGER DEFAULT 0,
health_score REAL DEFAULT 0,
credibility REAL DEFAULT 0.5,
last_seen_at TEXT,
added_at TEXT NOT NULL,
updated_at TEXT NOT NULL
)
""")
self._conn.execute("""
CREATE TABLE IF NOT EXISTS copy_signals (
id INTEGER PRIMARY KEY AUTOINCREMENT,
condition_id TEXT NOT NULL,
market_question TEXT,
market_slug TEXT DEFAULT '',
side TEXT NOT NULL,
outcome TEXT NOT NULL,
entry_price REAL NOT NULL,
aggregated_strength REAL NOT NULL,
source_wallets_json TEXT NOT NULL,
kelly_fraction REAL NOT NULL,
suggested_size_usd REAL NOT NULL,
generated_at TEXT NOT NULL,
executed INTEGER DEFAULT 0,
exit_price REAL,
pnl_usd REAL,
resolved_at TEXT
)
""")
self._conn.execute("""
CREATE INDEX IF NOT EXISTS idx_copy_signals_condition
ON copy_signals(condition_id)
""")
self._conn.execute("""
CREATE INDEX IF NOT EXISTS idx_copy_signals_generated_at
ON copy_signals(generated_at DESC)
""")
# Migration: add market_slug if missing (existing DBs)
try:
self._conn.execute(
"ALTER TABLE copy_signals ADD COLUMN market_slug TEXT DEFAULT ''"
)
except Exception:
pass
# Migration: add n_contributors if missing (existing DBs)
try:
self._conn.execute(
"ALTER TABLE copy_signals ADD COLUMN n_contributors INTEGER DEFAULT 0"
)
except Exception:
pass
# Backfill n_contributors from source_wallets_json for existing rows
try:
rows = self._conn.execute(
"SELECT id, source_wallets_json FROM copy_signals WHERE n_contributors = 0"
).fetchall()
for row in rows:
wallets = json.loads(row["source_wallets_json"])
if isinstance(wallets, list):
self._conn.execute(
"UPDATE copy_signals SET n_contributors = ? WHERE id = ?",
(len(wallets), row["id"]),
)
if rows:
logger.info(
f"[db] backfilled n_contributors for {len(rows)} existing signals"
)
except Exception:
pass
self._conn.execute("""
CREATE TABLE IF NOT EXISTS trade_executions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
signal_id INTEGER,
condition_id TEXT NOT NULL,
side TEXT NOT NULL,
outcome TEXT NOT NULL,
size_usd REAL NOT NULL,
price REAL NOT NULL,
order_id TEXT,
status TEXT NOT NULL,
error TEXT,
executed_at TEXT NOT NULL,
FOREIGN KEY(signal_id) REFERENCES copy_signals(id)
)
""")
self._conn.execute("""
CREATE TABLE IF NOT EXISTS pool_state (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
)
""")
# Credibility history (append-only; one row per Bayesian update)
self._conn.execute("""
CREATE TABLE IF NOT EXISTS credibility_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
address TEXT NOT NULL,
credibility REAL NOT NULL,
realized_pnl_window REAL,
n_closed INTEGER,
updated_at TEXT NOT NULL
)
""")
self._conn.execute("""
CREATE INDEX IF NOT EXISTS idx_cred_hist_address
ON credibility_history(address, updated_at)
""")
# Migration: add outcome_correct to copy_signals (existing DBs)
try:
self._conn.execute(
"ALTER TABLE copy_signals ADD COLUMN outcome_correct INTEGER"
)
except Exception:
pass
# ----- Wallet target operations -----
def upsert_wallet_target(self, wallet: dict) -> None:
now = datetime.now().isoformat()
with self._get_conn() as conn:
conn.execute("""
INSERT INTO wallet_targets
(address, source, pnl_30d_usd, pnl_total_usd, trades_count,
categories_count, health_score, credibility, last_seen_at,
added_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(address) DO UPDATE SET
pnl_30d_usd=excluded.pnl_30d_usd,
pnl_total_usd=excluded.pnl_total_usd,
trades_count=excluded.trades_count,
categories_count=excluded.categories_count,
health_score=excluded.health_score,
credibility=excluded.credibility,
last_seen_at=excluded.last_seen_at,
updated_at=excluded.updated_at
""", (
wallet["address"], wallet["source"],
wallet.get("pnl_30d_usd", 0),
wallet.get("pnl_total_usd", 0),
wallet.get("trades_count", 0),
wallet.get("categories_count", 0),
wallet.get("health_score", 0),
wallet.get("credibility", 0.5),
wallet.get("last_seen_at"),
wallet.get("added_at", now),
now,
))
now = _now_iso_utc()
with self._lock:
with self._conn:
self._conn.execute("""
INSERT INTO wallet_targets
(address, source, pnl_30d_usd, pnl_total_usd, trades_count,
categories_count, health_score, credibility, last_seen_at,
added_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(address) DO UPDATE SET
pnl_30d_usd=excluded.pnl_30d_usd,
pnl_total_usd=excluded.pnl_total_usd,
trades_count=excluded.trades_count,
categories_count=excluded.categories_count,
health_score=excluded.health_score,
credibility=excluded.credibility,
last_seen_at=excluded.last_seen_at,
updated_at=excluded.updated_at
""", (
wallet["address"], wallet["source"],
wallet.get("pnl_30d_usd", 0),
wallet.get("pnl_total_usd", 0),
wallet.get("trades_count", 0),
wallet.get("categories_count", 0),
wallet.get("health_score", 0),
wallet.get("credibility", 0.5),
wallet.get("last_seen_at"),
wallet.get("added_at", now),
now,
))
def get_all_wallet_targets(self) -> List[dict]:
with self._get_conn() as conn:
rows = conn.execute(
with self._lock:
rows = self._conn.execute(
"SELECT * FROM wallet_targets ORDER BY health_score DESC"
).fetchall()
return [dict(r) for r in rows]
def get_wallet_addresses(self) -> List[str]:
with self._get_conn() as conn:
rows = conn.execute("SELECT address FROM wallet_targets").fetchall()
with self._lock:
rows = self._conn.execute(
"SELECT address FROM wallet_targets"
).fetchall()
return [r["address"] for r in rows]
def delete_wallet_targets_not_in(self, keep_addresses: List[str]) -> int:
"""Remove wallets whose address is not in keep_addresses. Returns deleted count."""
if not keep_addresses:
return 0
placeholders = ",".join("?" for _ in keep_addresses)
with self._lock:
with self._conn:
cur = self._conn.execute(
f"DELETE FROM wallet_targets WHERE address NOT IN ({placeholders})",
keep_addresses,
)
if cur.rowcount:
logger.info(f"[db] pruned {cur.rowcount} stale wallets")
return cur.rowcount
def get_wallet_target(self, address: str) -> Optional[dict]:
with self._get_conn() as conn:
row = conn.execute(
with self._lock:
row = self._conn.execute(
"SELECT * FROM wallet_targets WHERE address=?", (address,)
).fetchone()
return dict(row) if row else None
def update_wallet_credibility(self, address: str, credibility: float) -> None:
with self._get_conn() as conn:
conn.execute(
"UPDATE wallet_targets SET credibility=?, updated_at=? WHERE address=?",
(credibility, datetime.now().isoformat(), address),
)
"""Update wallet credibility AND append to credibility_history.
Previously this was a pure UPDATE that overwrote the prior value,
making it impossible to audit how credibility evolved. Now every
update also inserts a row into ``credibility_history`` so we can
plot the curve and validate the Bayesian formula.
"""
now = _now_iso_utc()
with self._lock:
with self._conn:
self._conn.execute(
"UPDATE wallet_targets SET credibility=?, updated_at=? WHERE address=?",
(credibility, now, address),
)
self._conn.execute("""
INSERT INTO credibility_history
(address, credibility, realized_pnl_window, n_closed, updated_at)
VALUES (?, ?, NULL, NULL, ?)
""", (address, credibility, now))
def update_wallet_credibility_with_context(
self,
address: str,
credibility: float,
realized_pnl_window: Optional[float] = None,
n_closed: Optional[int] = None,
) -> None:
"""Same as update_wallet_credibility but records PnL context in history."""
now = _now_iso_utc()
with self._lock:
with self._conn:
self._conn.execute(
"UPDATE wallet_targets SET credibility=?, updated_at=? WHERE address=?",
(credibility, now, address),
)
self._conn.execute("""
INSERT INTO credibility_history
(address, credibility, realized_pnl_window, n_closed, updated_at)
VALUES (?, ?, ?, ?, ?)
""", (address, credibility, realized_pnl_window, n_closed, now))
def get_credibility_history(
self, address: str, limit: int = 200,
) -> List[dict]:
with self._lock:
rows = self._conn.execute(
"SELECT * FROM credibility_history WHERE address=? "
"ORDER BY updated_at ASC LIMIT ?",
(address, limit),
).fetchall()
return [dict(r) for r in rows]
def get_signals_count(self) -> int:
with self._get_conn() as conn:
return conn.execute("SELECT COUNT(*) FROM copy_signals").fetchone()[0]
with self._lock:
return self._conn.execute(
"SELECT COUNT(*) FROM copy_signals"
).fetchone()[0]
def get_wallet_count(self) -> int:
with self._get_conn() as conn:
return conn.execute("SELECT COUNT(*) FROM wallet_targets").fetchone()[0]
with self._lock:
return self._conn.execute(
"SELECT COUNT(*) FROM wallet_targets"
).fetchone()[0]
def get_executed_count(self) -> int:
with self._get_conn() as conn:
return conn.execute(
with self._lock:
return self._conn.execute(
"SELECT COUNT(*) FROM copy_signals WHERE executed=1"
).fetchone()[0]
# ----- Copy signal operations -----
def insert_signal(self, signal: dict) -> int:
with self._get_conn() as conn:
cur = conn.execute("""
INSERT INTO copy_signals
(condition_id, market_question, side, outcome, entry_price,
aggregated_strength, source_wallets_json, kelly_fraction,
suggested_size_usd, generated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
signal["condition_id"],
signal.get("market_question", ""),
signal["side"],
signal["outcome"],
signal["entry_price"],
signal["aggregated_strength"],
json.dumps(signal["source_wallets"]),
signal["kelly_fraction"],
signal["suggested_size_usd"],
datetime.now().isoformat(),
))
return cur.lastrowid
with self._lock:
with self._conn:
cur = self._conn.execute("""
INSERT INTO copy_signals
(condition_id, market_question, market_slug, side, outcome, entry_price,
aggregated_strength, source_wallets_json, kelly_fraction,
suggested_size_usd, n_contributors, generated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
signal["condition_id"],
signal.get("market_question", ""),
signal.get("market_slug", ""),
signal["side"],
signal["outcome"],
signal["entry_price"],
signal["aggregated_strength"],
json.dumps(signal["source_wallets"]),
signal["kelly_fraction"],
signal["suggested_size_usd"],
signal.get("n_contributors", 0),
_now_iso_utc(),
))
return cur.lastrowid
def mark_signal_executed(self, signal_id: int) -> None:
with self._get_conn() as conn:
conn.execute(
"UPDATE copy_signals SET executed=1 WHERE id=?", (signal_id,)
)
with self._lock:
with self._conn:
self._conn.execute(
"UPDATE copy_signals SET executed=1 WHERE id=?", (signal_id,)
)
def get_unresolved_signals(self, limit: int = 50) -> List[dict]:
"""Signals whose market hasn't been backfilled yet."""
with self._lock:
rows = self._conn.execute(
"SELECT id, condition_id, outcome, entry_price, suggested_size_usd "
"FROM copy_signals WHERE resolved_at IS NULL "
"ORDER BY generated_at ASC LIMIT ?",
(limit,),
).fetchall()
return [dict(r) for r in rows]
def update_signal_outcome(
self,
signal_id: int,
exit_price: float,
pnl_usd: float,
outcome_correct: int,
resolved_at: str,
) -> None:
with self._lock:
with self._conn:
self._conn.execute(
"UPDATE copy_signals SET exit_price=?, pnl_usd=?, "
"outcome_correct=?, resolved_at=? WHERE id=?",
(exit_price, pnl_usd, outcome_correct, resolved_at, signal_id),
)
def mark_signal_resolved_unpnl(self, signal_id: int, resolved_at: str) -> None:
"""Mark signal as resolved without PnL (e.g. cancelled market)."""
with self._lock:
with self._conn:
self._conn.execute(
"UPDATE copy_signals SET resolved_at=?, outcome_correct=-1 "
"WHERE id=?",
(resolved_at, signal_id),
)
def get_recent_signals(self, limit: int = 50) -> List[dict]:
with self._get_conn() as conn:
rows = conn.execute(
with self._lock:
rows = self._conn.execute(
"SELECT * FROM copy_signals ORDER BY generated_at DESC LIMIT ?",
(limit,),
).fetchall()
return [dict(r) for r in rows]
def record_execution(self, exec_row: dict) -> int:
with self._get_conn() as conn:
cur = conn.execute("""
INSERT INTO trade_executions
(signal_id, condition_id, side, outcome, size_usd, price,
order_id, status, error, executed_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
exec_row.get("signal_id"),
exec_row["condition_id"],
exec_row["side"],
exec_row["outcome"],
exec_row["size_usd"],
exec_row["price"],
exec_row.get("order_id"),
exec_row["status"],
exec_row.get("error"),
datetime.now().isoformat(),
))
return cur.lastrowid
with self._lock:
with self._conn:
cur = self._conn.execute("""
INSERT INTO trade_executions
(signal_id, condition_id, side, outcome, size_usd, price,
order_id, status, error, executed_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
exec_row.get("signal_id"),
exec_row["condition_id"],
exec_row["side"],
exec_row["outcome"],
exec_row["size_usd"],
exec_row["price"],
exec_row.get("order_id"),
exec_row["status"],
exec_row.get("error"),
_now_iso_utc(),
))
return cur.lastrowid
# ----- Stats -----
def get_stats(self) -> dict:
with self._get_conn() as conn:
total = conn.execute("SELECT COUNT(*) FROM wallet_targets").fetchone()[0]
sigs = conn.execute("SELECT COUNT(*) FROM copy_signals").fetchone()[0]
executed = conn.execute(
with self._lock:
total = self._conn.execute(
"SELECT COUNT(*) FROM wallet_targets"
).fetchone()[0]
sigs = self._conn.execute(
"SELECT COUNT(*) FROM copy_signals"
).fetchone()[0]
executed = self._conn.execute(
"SELECT COUNT(*) FROM copy_signals WHERE executed=1"
).fetchone()[0]
return {
@@ -227,3 +457,177 @@ class CopyTraderDatabase:
"total_signals": sigs,
"executed_signals": executed,
}
def get_outcome_stats(self) -> dict:
"""Overall signal outcome statistics for strategy evaluation."""
with self._lock:
row = self._conn.execute("""
SELECT
COUNT(*) AS total,
SUM(CASE WHEN outcome_correct = 1 THEN 1 ELSE 0 END) AS wins,
SUM(CASE WHEN outcome_correct = 0 THEN 1 ELSE 0 END) AS losses,
SUM(CASE WHEN outcome_correct IS NULL AND resolved_at IS NULL THEN 1 ELSE 0 END) AS pending,
SUM(CASE WHEN outcome_correct = -1 THEN 1 ELSE 0 END) AS cancelled,
COALESCE(SUM(pnl_usd), 0) AS total_pnl,
COALESCE(AVG(CASE WHEN outcome_correct IN (0,1) THEN pnl_usd END), 0) AS avg_pnl,
COALESCE(MAX(pnl_usd), 0) AS best_pnl,
COALESCE(MIN(pnl_usd), 0) AS worst_pnl
FROM copy_signals
""").fetchone()
total = row["total"] or 0
wins = row["wins"] or 0
losses = row["losses"] or 0
decided = wins + losses
return {
"total_signals": total,
"wins": wins,
"losses": losses,
"pending": row["pending"] or 0,
"cancelled": row["cancelled"] or 0,
"hit_rate": (wins / decided) if decided > 0 else 0.0,
"total_pnl_usd": float(row["total_pnl"] or 0),
"avg_pnl_usd": float(row["avg_pnl"] or 0),
"best_pnl_usd": float(row["best_pnl"] or 0),
"worst_pnl_usd": float(row["worst_pnl"] or 0),
}
def get_signal_stats_by_strength(self) -> List[dict]:
"""Bucket resolved signals by aggregated_strength to find which
strength ranges actually predict winning outcomes.
Used to calibrate CONSENSUS_STRENGTH_THRESHOLD and the Kelly sizer's
empirical hit-rate lookup. Bucket ranges accommodate the current
sum(cred)*log1p(n) formula which can exceed 1.0 for strong consensus.
"""
with self._lock:
rows = self._conn.execute("""
SELECT
CASE
WHEN aggregated_strength < 0.5 THEN '0.0-0.5'
WHEN aggregated_strength < 1.0 THEN '0.5-1.0'
WHEN aggregated_strength < 2.0 THEN '1.0-2.0'
WHEN aggregated_strength < 3.0 THEN '2.0-3.0'
WHEN aggregated_strength < 5.0 THEN '3.0-5.0'
ELSE '5.0+'
END AS bucket,
COUNT(*) AS n,
SUM(CASE WHEN outcome_correct = 1 THEN 1 ELSE 0 END) AS wins,
COALESCE(SUM(pnl_usd), 0) AS pnl
FROM copy_signals
WHERE outcome_correct IN (0, 1)
GROUP BY bucket
ORDER BY bucket
""").fetchall()
return [dict(r) for r in rows]
def get_signal_stats_by_wallet_count(self) -> List[dict]:
"""Bucket by n_contributors to calibrate CONSENSUS_MIN_WALLETS."""
with self._lock:
rows = self._conn.execute("""
SELECT
n_contributors AS bucket,
COUNT(*) AS n,
SUM(CASE WHEN outcome_correct = 1 THEN 1 ELSE 0 END) AS wins,
COALESCE(SUM(pnl_usd), 0) AS pnl
FROM copy_signals
WHERE outcome_correct IN (0, 1) AND n_contributors > 0
GROUP BY n_contributors
ORDER BY n_contributors
""").fetchall()
return [dict(r) for r in rows]
def get_signal_stats_by_wallet(self, limit: int = 30) -> List[dict]:
"""Per-wallet hit rate based on its signals in source_wallets_json.
Useful for reverse-validating the credibility formula: do high-
credibility wallets actually win more often?
"""
with self._lock:
rows = self._conn.execute("""
SELECT
json_each.value AS address,
COUNT(*) AS n_signals,
SUM(CASE WHEN s.outcome_correct = 1 THEN 1 ELSE 0 END) AS wins,
COALESCE(SUM(s.pnl_usd), 0) AS pnl
FROM copy_signals s, json_each(s.source_wallets_json)
WHERE s.outcome_correct IN (0, 1)
GROUP BY json_each.value
ORDER BY pnl DESC
LIMIT ?
""", (limit,)).fetchall()
return [dict(r) for r in rows]
def get_signal_stats_by_category(self, limit: int = 30) -> List[dict]:
"""Per-category hit rate. Uses market_slug prefix as a proxy for
category (Polymarket slugs often encode the category, e.g.
'crypto-...')."""
with self._lock:
rows = self._conn.execute("""
SELECT
CASE
WHEN market_slug LIKE 'crypto%' THEN 'crypto'
WHEN market_slug LIKE 'politics%' THEN 'politics'
WHEN market_slug LIKE 'sports%' THEN 'sports'
WHEN market_slug LIKE 'entertainment%' THEN 'entertainment'
WHEN market_slug LIKE 'world%' THEN 'world'
ELSE 'other'
END AS category,
COUNT(*) AS n,
SUM(CASE WHEN outcome_correct = 1 THEN 1 ELSE 0 END) AS wins,
COALESCE(SUM(pnl_usd), 0) AS pnl
FROM copy_signals
WHERE outcome_correct IN (0, 1)
GROUP BY category
ORDER BY n DESC
LIMIT ?
""", (limit,)).fetchall()
return [dict(r) for r in rows]
def get_signal_stats_by_hour(self) -> List[dict]:
"""Per-hour-of-day signal volume + hit rate.
Identifies active/quiet hours and whether signals in certain hours
perform better. Useful for scheduling the pool refresh.
"""
with self._lock:
rows = self._conn.execute("""
SELECT
CAST(strftime('%H', generated_at) AS INTEGER) AS hour,
COUNT(*) AS n,
SUM(CASE WHEN outcome_correct = 1 THEN 1 ELSE 0 END) AS wins,
COALESCE(SUM(pnl_usd), 0) AS pnl
FROM copy_signals
WHERE outcome_correct IN (0, 1)
GROUP BY hour
ORDER BY hour
""").fetchall()
return [dict(r) for r in rows]
# ----- Pool state (persistent scheduling) -----
def get_pool_state(self, key: str) -> Optional[str]:
with self._lock:
row = self._conn.execute(
"SELECT value FROM pool_state WHERE key=?", (key,)
).fetchone()
return row["value"] if row else None
def set_pool_state(self, key: str, value: str) -> None:
with self._lock:
with self._conn:
self._conn.execute(
"INSERT INTO pool_state(key, value) VALUES(?, ?) "
"ON CONFLICT(key) DO UPDATE SET value=excluded.value",
(key, value),
)
def get_pool_state_json(self, key: str) -> Optional[dict]:
raw = self.get_pool_state(key)
if raw is None:
return None
try:
return json.loads(raw)
except (ValueError, TypeError):
return None
def set_pool_state_json(self, key: str, value: dict) -> None:
self.set_pool_state(key, json.dumps(value, default=str))
src/main.py
+293 -76
View File
@@ -10,7 +10,8 @@ Commands:
import asyncio
import signal as sys_signal
import sys
from datetime import datetime
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Optional
@@ -21,12 +22,14 @@ from src.db.database import CopyTraderDatabase
from src.services.aggregator import SignalAggregator
from src.services.bayesian import BayesianUpdater
from src.services.kelly import KellySizer
from src.services.risk_manager import RiskManager
from src.services.telegram import TelegramNotifier
from src.services.user_stream import UserTradeStream
from src.services.wallet_pool import WalletPoolBuilder
from src.services.outcome_resolver import OutcomeResolver
from src.utils.logger import BotLogger, setup_logging
app = typer.Typer(help="Polymarket Copy Trader — follow top wallets via consensus")
app = typer.Typer(help="Polymarket 跟单交易 — 通过共识机制跟随优质钱包")
logger = BotLogger()
@@ -36,12 +39,19 @@ class CopyTrader:
def __init__(self):
self.settings = get_settings()
self.db = CopyTraderDatabase(self.settings.db_path)
self.sizer = KellySizer()
# KellySizer needs DB access to look up empirical hit-rates by
# strength bucket. Without DB it falls back to the conservative
# fixed fraction (KELLY_FALLBACK_FRACTION).
self.sizer = KellySizer(db=self.db)
self.risk = RiskManager(self.db)
self.pool_builder = WalletPoolBuilder(self.db)
self.aggregator = SignalAggregator(self.db, sizer=self.sizer)
self.aggregator = SignalAggregator(
self.db, sizer=self.sizer, risk_manager=self.risk
)
self.bayes = BayesianUpdater(self.db, self.aggregator)
self.stream = UserTradeStream(self.db)
self.notifier = TelegramNotifier()
self.resolver = OutcomeResolver(self.db)
self._running = False
self._tasks = []
@@ -55,97 +65,272 @@ class CopyTrader:
capital=self.settings.initial_capital_usd,
)
logger.info(
f"Settings: min_trade=${self.settings.min_trade_size_usd} "
f"price=[{self.settings.min_price}, {self.settings.max_price}] "
f"poll={self.settings.user_poll_interval_seconds}s "
f"consensus_wallets={self.settings.consensus_min_wallets}"
f"设置：最小交易=${self.settings.min_trade_size_usd} "
f"价格=[{self.settings.min_price}, {self.settings.max_price}] "
f"轮询={self.settings.user_poll_interval_seconds}s "
f"共识钱包={self.settings.consensus_min_wallets}"
)
loop = asyncio.get_running_loop()
self._loop_coroutines = {
"pool-refresh": self._pool_refresh_loop,
"trade-stream": self._trade_stream_loop,
"credibility": self._credibility_loop,
"acc-cleanup": self._accumulator_cleanup_loop,
"health": self._health_beat,
"outcome-resolve": self._outcome_resolve_loop,
}
def _on_done(name: str, t: asyncio.Task) -> None:
self._on_task_done(name, t, loop)
self._tasks = [
asyncio.create_task(self._pool_refresh_loop(), name="pool-refresh"),
asyncio.create_task(self._trade_stream_loop(), name="trade-stream"),
asyncio.create_task(self._credibility_loop(), name="credibility"),
asyncio.create_task(self._health_beat(), name="health"),
loop.create_task(coro(), name=name)
for name, coro in self._loop_coroutines.items()
]
for name, t in zip(self._loop_coroutines.keys(), self._tasks):
t.add_done_callback(lambda t, n=name: _on_done(n, t))
try:
await asyncio.gather(*self._tasks)
await asyncio.gather(*self._tasks, return_exceptions=True)
except asyncio.CancelledError:
logger.info("Cancelled")
logger.info("已取消")
finally:
await self._teardown()
async def _teardown(self) -> None:
"""Flush in-memory state to DB and close resources on shutdown."""
try:
self.aggregator.flush_accumulators()
self.aggregator._save_debounce()
logger.info("[退出] 已 flush 累加器与去重状态")
except Exception as e:
logger.error(f"[退出] flush 状态失败：{e}")
try:
await self.notifier.close()
except Exception:
pass
try:
self.db.close()
logger.info("[退出] DB 已关闭")
except Exception as e:
logger.error(f"[退出] 关闭 DB 失败：{e}")
def _on_task_done(self, name: str, t: asyncio.Task, loop: asyncio.AbstractEventLoop) -> None:
"""Diagnose unexpected task death; respawn if still running."""
if not self._running:
return
if t.cancelled():
logger.warning(f"[循环] 任务 '{name}' 被取消，尝试恢复")
else:
exc = t.exception()
if exc is None:
logger.warning(f"[循环] 任务 '{name}' 正常结束但未预期，尝试恢复")
else:
logger.error(
f"[循环] 任务 '{name}' 崩溃：{type(exc).__name__}: {exc}",
exc_info=exc,
)
coro_factory = self._loop_coroutines.get(name)
if coro_factory is None or not self._running:
return
try:
new_t = loop.create_task(coro_factory(), name=name)
new_t.add_done_callback(lambda nt, n=name: self._on_task_done(n, nt, loop))
self._tasks = [x if x is not t else new_t for x in self._tasks]
logger.info(f"[循环] 任务 '{name}' 已重建")
except Exception as e:
logger.error(f"[循环] 重建任务 '{name}' 失败：{e}")
async def _bootstrap(self) -> None:
logger.info("Bootstrapping...")
logger.info("初始化中...")
await self.notifier.start()
self.aggregator.load_credibilities()
if self.aggregator.wallet_credibility == {}:
logger.info("No wallets in pool — running initial pool build")
logger.info("钱包池为空，正在首次构建钱包池")
try:
pool = await self.pool_builder.build_pool_async()
now = datetime.now().isoformat()
now = datetime.now(timezone.utc).isoformat()
for w in pool:
w.setdefault("added_at", now)
self.db.upsert_wallet_target(w)
logger.info(f"Initial pool built: {len(pool)} wallets")
self.db.set_pool_state("last_pool_refresh", str(time.time()))
logger.info(f"钱包池首次构建完成：{len(pool)} 个钱包")
except Exception as e:
logger.error(f"Initial pool build failed: {e}")
await self.notifier.send_error(f"Initial pool build failed: {e}")
logger.error(f"钱包池首次构建失败：{e}")
await self.notifier.send_error(f"钱包池初始化失败：{e}")
self.aggregator.load_credibilities()
logger.info(f"Pool loaded: {len(self.aggregator.wallet_credibility)} wallets")
elif self.db.get_pool_state("last_pool_refresh") is None:
self.db.set_pool_state("last_pool_refresh", str(time.time()))
logger.info("钱包池存在但无历史刷新时间戳，标记为当前")
logger.info(f"钱包池加载完成：{len(self.aggregator.wallet_credibility)} 个钱包")
async def _pool_refresh_loop(self) -> None:
seconds = self.settings.wallet_refresh_hours * 3600
logger.info(f"[loop] pool refresh every {seconds}s, first refresh after {seconds}s")
await asyncio.sleep(seconds)
while self._running:
try:
pool = await self.pool_builder.build_pool_async()
now = datetime.now().isoformat()
for w in pool:
w.setdefault("added_at", now)
self.db.upsert_wallet_target(w)
logger.info(f"[loop] pool refreshed: {len(pool)} wallets")
self.aggregator.load_credibilities()
await self.notifier.send_signal(
market="Pool refreshed",
side="INFO",
strength=0.0,
size_usd=0,
n_wallets=len(pool),
interval = self.settings.wallet_refresh_hours * 3600
last_refresh = self._read_last_pool_refresh()
now_ts = time.time()
if last_refresh is None:
sleep_for = interval
logger.info(
f"[循环] 钱包池无历史刷新记录，{sleep_for}s 后首次刷新"
)
else:
elapsed = now_ts - last_refresh
remaining = max(0, interval - elapsed)
if remaining == 0:
logger.info(
f"[循环] 钱包池已过 {elapsed/3600:.1f}h 超过 {interval/3600:.0f}h 间隔，"
f"立即刷新"
)
except Exception as e:
logger.error(f"[loop] pool refresh failed: {e}")
await self.notifier.send_error(f"Pool refresh failed: {e}")
await asyncio.sleep(seconds)
else:
logger.info(
f"[循环] 钱包池上次刷新于 {elapsed/3600:.1f}h 前，"
f"下次刷新在 {remaining/3600:.1f}h 后（每 {interval/3600:.0f}h 一次）"
)
sleep_for = remaining
await asyncio.sleep(sleep_for)
while self._running:
await self._refresh_pool_once()
await asyncio.sleep(interval)
async def _refresh_pool_once(self) -> None:
try:
old_addresses = set(self.db.get_wallet_addresses())
pool = await self.pool_builder.build_pool_async()
now_iso = datetime.now(timezone.utc).isoformat()
now_ts = time.time()
new_addresses = {w["address"] for w in pool}
for w in pool:
w.setdefault("added_at", now_iso)
self.db.upsert_wallet_target(w)
self.db.set_pool_state("last_pool_refresh", str(now_ts))
removed = old_addresses - new_addresses
if removed:
self.db.delete_wallet_targets_not_in(list(new_addresses))
for addr in removed:
self.aggregator.purge_wallet(addr)
logger.info(f"[循环] 移除 {len(removed)} 个掉队钱包")
logger.info(f"[循环] 钱包池已刷新：{len(pool)} 个钱包")
self.aggregator.load_credibilities()
await self.notifier.send_signal(
market="钱包池已刷新",
side="INFO",
strength=0.0,
size_usd=0,
n_wallets=len(pool),
)
except Exception as e:
logger.error(f"[循环] 钱包池刷新失败：{e}")
await self.notifier.send_error(f"钱包池刷新失败：{e}")
def _read_last_pool_refresh(self) -> Optional[float]:
raw = self.db.get_pool_state("last_pool_refresh")
if raw:
try:
return float(raw)
except ValueError:
return None
return None
async def _trade_stream_loop(self) -> None:
logger.info("[loop] trade stream starting")
logger.info("[循环] 交易流启动中")
async def on_trade(addr: str, trade: dict) -> None:
signal = await self.aggregator.on_trade(addr, trade)
if signal:
sig_id = self.db.insert_signal(signal)
logger.info(
f"[signal] #{sig_id} stored: {signal['side']} ${signal['suggested_size_usd']:.0f} "
f"on {signal['market_question'][:40]}"
f"[信号] #{sig_id} 已入库：{signal['side']} ${signal['suggested_size_usd']:.0f} "
f"市场：{signal['market_question'][:40]}"
)
await self.notifier.send_signal(
market=signal["market_question"],
side=f"{signal['side']} {signal['outcome']}",
strength=signal["aggregated_strength"],
size_usd=signal["suggested_size_usd"],
n_wallets=signal["n_contributors"],
# Fire-and-forget: slow Telegram sends must not block the
# aggregator from processing subsequent trades.
asyncio.create_task(
self.notifier.send_signal(
market=signal["market_question"],
side=f"{signal['side']}",
strength=signal["aggregated_strength"],
size_usd=signal["suggested_size_usd"],
n_wallets=signal["n_contributors"],
condition_id=signal["condition_id"],
outcome=signal["outcome"],
entry_price=signal["entry_price"],
market_slug=signal.get("market_slug", ""),
)
)
await self.stream.run(on_trade)
async def _credibility_loop(self) -> None:
seconds = self.settings.credibility_update_minutes * 60
logger.info(f"[loop] credibility update every {seconds}s")
interval = self.settings.credibility_update_minutes * 60
raw = self.db.get_pool_state("last_credibility_update")
now_ts = time.time()
if raw is None:
sleep_for = interval
logger.info(f"[循环] 信誉分无历史更新记录，{sleep_for}s 后首次更新")
else:
try:
last_ts = float(raw)
except ValueError:
last_ts = now_ts
elapsed = now_ts - last_ts
remaining = max(0, interval - elapsed)
if remaining == 0:
logger.info(
f"[循环] 信誉分已过 {elapsed/60:.1f}min 超过 {interval/60:.0f}min 间隔，立即更新"
)
else:
logger.info(
f"[循环] 信誉分上次更新于 {elapsed/60:.1f}min 前，"
f"下次更新在 {remaining/60:.1f}min 后（每 {interval/60:.0f}min 一次）"
)
sleep_for = remaining
await asyncio.sleep(sleep_for)
while self._running:
await asyncio.sleep(seconds)
try:
await self.bayes._update_once()
self.db.set_pool_state("last_credibility_update", str(time.time()))
except Exception as e:
logger.error(f"[loop] credibility update failed: {e}")
logger.error(f"[循环] 信誉分更新失败：{e}")
await asyncio.sleep(interval)
async def _outcome_resolve_loop(self) -> None:
"""Periodically backfill resolved-market outcomes onto copy_signals."""
interval = self.settings.outcome_resolve_minutes * 60
logger.info(f"[循环] 信号结果回填启动 (每 {self.settings.outcome_resolve_minutes}min)")
# Short initial delay so the bot fully boots before first resolve.
await asyncio.sleep(60)
while self._running:
try:
result = await self.resolver.resolve_once()
if result["resolved"] > 0:
logger.info(
f"[循环] 信号回填：{result['resolved']} 新结果 / "
f"{result['pending'] if 'pending' in result else result['skipped']} 待定 / "
f"{result['failed']} 失败"
)
except Exception as e:
logger.error(f"[循环] 信号结果回填失败：{e}")
await asyncio.sleep(interval)
async def _accumulator_cleanup_loop(self) -> None:
"""Periodically sweep expired market accumulators and prune stale _last_seen."""
interval = max(60, self.settings.consensus_window_seconds // 4)
while self._running:
await asyncio.sleep(interval)
try:
before = len(self.aggregator.accumulators)
self.aggregator._cleanup_expired()
self.stream.prune_last_seen(self.db.get_wallet_addresses())
after = len(self.aggregator.accumulators)
if before != after:
logger.info(
f"[循环] 清理 {before - after} 个过期累加器，剩余 {after}"
)
except Exception as e:
logger.error(f"[循环] 累加器清理失败：{e}")
async def _health_beat(self) -> None:
"""Emit periodic stats for log visibility."""
@@ -154,16 +339,15 @@ class CopyTrader:
stats = self.db.get_stats()
n_acc = len(self.aggregator.accumulators)
logger.info(
f"[health] wallets={stats['wallet_count']} "
f"signals={stats['total_signals']} "
f"executed={stats['executed_signals']} "
f"open_accumulators={n_acc}"
f"[心跳] 钱包数={stats['wallet_count']} "
f"信号数={stats['total_signals']} "
f"已执行={stats['executed_signals']} "
f"开放累加器={n_acc}"
)
def stop(self) -> None:
self._running = False
self.stream.stop()
self.bayes.stop()
for t in self._tasks:
t.cancel()
@@ -179,9 +363,9 @@ def signal_handler(signum, frame):
@app.command()
def run(
debug: bool = typer.Option(False, "--debug", "-d", help="Verbose DEBUG logs"),
debug: bool = typer.Option(False, "--debug", "-d", help="显示详细 DEBUG 日志"),
):
"""Start the copy-trader bot."""
"""启动跟单交易机器人。"""
global _watcher
setup_logging("DEBUG" if debug else "INFO")
@@ -199,30 +383,63 @@ def run(
@app.command()
def pool():
"""Rebuild the top-wallet pool now and exit."""
"""立即重建钱包池并退出。"""
setup_logging("INFO")
settings = get_settings()
db = CopyTraderDatabase(settings.db_path)
builder = WalletPoolBuilder(db)
count = builder.refresh()
print(f"Pool refreshed: {count} wallets persisted to {settings.db_path}")
print(f"钱包池已刷新：{count} 个钱包已写入 {settings.db_path}")
@app.command()
def stats():
"""Show database stats."""
"""显示数据库统计。"""
setup_logging("INFO")
settings = get_settings()
db = CopyTraderDatabase(settings.db_path)
s = db.get_stats()
print(f"Wallet count: {s['wallet_count']}")
print(f"Total signals: {s['total_signals']}")
print(f"Executed: {s['executed_signals']}")
o = db.get_outcome_stats()
print(f"钱包数： {s['wallet_count']}")
print(f"信号总数： {s['total_signals']}")
print(f"已执行： {s['executed_signals']}")
print(f"已 resolve： {o['wins'] + o['losses']}/{o['total_signals']}")
print(f" 胜 / 负： {o['wins']} / {o['losses']}")
print(f" 命中率： {o['hit_rate']*100:.1f}%")
print(f" 总 PnL： ${o['total_pnl_usd']:.2f}")
print(f" 平均 PnL： ${o['avg_pnl_usd']:.2f}")
@app.command()
def backfill(
batch: int = typer.Option(500, "--batch", "-b", help="单次最多处理的信号数"),
hours: int = typer.Option(0, "--hours", "-h", help="仅回填最近 N 小时内的信号(0=全部)"),
):
"""手动回填已 resolve 市场的信号结果。"""
setup_logging("INFO")
settings = get_settings()
db = CopyTraderDatabase(settings.db_path)
resolver = OutcomeResolver(db)
async def _run():
if hours > 0:
# Override batch size for manual runs
return await resolver.resolve_once(max_signals=batch)
return await resolver.resolve_once(max_signals=batch)
result = asyncio.run(_run())
print(
f"回填完成：扫描 {result['scanned']} / "
f"resolve {result['resolved']} / "
f"待定 {result['skipped']} / "
f"失败 {result['failed']}"
)
db.close()
@app.command()
def test_stream(minutes: int = 2):
"""One-shot test: poll top 3 wallets for N minutes, print new trades."""
"""一次性测试：轮询前 3 个钱包 N 分钟，打印新交易。"""
setup_logging("DEBUG")
async def _run():
settings = get_settings()
@@ -232,7 +449,7 @@ def test_stream(minutes: int = 2):
agg.load_credibilities()
stream = UserTradeStream(db)
addrs = db.get_wallet_addresses()[:3]
print(f"Top 3 wallets: {addrs}")
print(f"前 3 个钱包：{addrs}")
async def cb(addr, t):
print(
@@ -247,21 +464,21 @@ def test_stream(minutes: int = 2):
async def patched_run(on_trade):
self_ref = stream
self_ref._running = True
warmup_until = __import__('time').time() + settings.stream_warmup_seconds
warmup_until = time.time() + settings.stream_warmup_seconds
while self_ref._running:
for addr in addrs:
try:
trades = await asyncio.to_thread(self_ref.data.get_activity, addr, settings.stream_max_trades_per_wallet)
await self_ref._process_trades(addr, trades, on_trade, __import__('time').time() < warmup_until)
await self_ref._process_trades(addr, trades, on_trade, time.time() < warmup_until)
except Exception as e:
print(f"poll err: {e}")
print(f"轮询错误：{e}")
await asyncio.sleep(settings.user_poll_interval_seconds)
stream.run = patched_run
try:
await asyncio.wait_for(stream.run(cb), timeout=minutes * 60)
except asyncio.TimeoutError:
print(f"Test stream ended after {minutes} min")
print(f"测试流已结束，时长 {minutes} 分钟")
asyncio.run(_run())
@@ -272,7 +489,7 @@ def dashboard(port: int = 8518, host: str = "0.0.0.0"):
import uvicorn
from src.dashboard import app as dashboard_app
print(f"Dashboard starting at http://{host}:{port}")
print(f"Dashboard 启动中，访问 http://{host}:{port}")
uvicorn.run(dashboard_app, host=host, port=port, log_level="info")
src/models/wallet.py
+2 -2
View File
@@ -1,6 +1,6 @@
"""Data models for the copy-trader."""
from dataclasses import dataclass, field
from datetime import datetime
from datetime import datetime, timezone
from typing import Optional
@@ -31,4 +31,4 @@ class CopySignal:
source_wallets: list # list of (address, size_usd, weight)
kelly_fraction: float
suggested_size_usd: float
generated_at: str = field(default_factory=lambda: datetime.now().isoformat())
generated_at: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
src/services/aggregator.py
+453 -60
View File
@@ -13,27 +13,114 @@ Flow:
Consensus rule:
- N wallets (≥ CONSENSUS_MIN_WALLETS) agree on same direction within window
- aggregated strength = mean credibility of agreeing wallets
- aggregated strength = sum(credibility) * log1p(n) — rewards both count
and credibility with diminishing returns on count
- emit only when (agreed_strength - opposed_strength) >= CONSENSUS_STRENGTH_THRESHOLD
Signal filters (in on_trade):
- SELL trades are logged but NOT emitted when BUY_ONLY_SIGNALS=True (we
cannot follow SELLs without position tracking; SELL on Polymarket =
selling tokens you already hold, not shorting).
- Same-market cooldown: once a signal fires for a condition_id, suppress
new signals on that market for SIGNAL_COOLDOWN_SECONDS.
- min-hours-to-resolution: skip markets resolving within
MIN_HOURS_TO_RESOLUTION (thin liquidity + resolution risk).
- Add-position handling: if a wallet already contributed to an
accumulator, subsequent trades from it bump size_usd but do NOT
re-add credibility strength (prevents one wallet inflating consensus).
Restart-safety
--------------
``wallet_debounce`` (per wallet+market last-trade time) is persisted to
``pool_state`` so that a restart does not let the same wallet re-trigger
the same market immediately. Stale entries (older than the debounce window)
are filtered out on load. ``accumulators`` are also persisted (existing
behavior) and ``_persist_accumulators`` is throttled to avoid writing the
full accumulator snapshot on every single trade.
"""
import asyncio
import json
import logging
import math
import time
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Optional
from datetime import datetime, timezone
from typing import Dict, List, Optional, Tuple
from src.config import get_settings
from src.db.database import CopyTraderDatabase
from src.services.data_api import DataAPIClient
from src.services.data_api import DataAPIClient, get_shared_data_client
from src.services.kelly import KellySizer
logger = logging.getLogger(__name__)
_DEBOUNCE_STATE_KEY = "wallet_debounce"
_PERSIST_DEBOUNCE_SECONDS = 5.0 # throttle accumulator snapshots
def _parse_end_date(raw) -> float:
"""Parse Polymarket /markets `endDate` (ISO 8601) → epoch seconds.
Returns 0.0 on missing/invalid input. Examples observed in the wild:
'2025-11-04T23:59:59Z', '2025-12-31T00:00:00.000Z'.
"""
if not raw or not isinstance(raw, str):
return 0.0
try:
# datetime.fromisoformat doesn't accept 'Z' suffix pre-3.11
s = raw.replace("Z", "+00:00")
dt = datetime.fromisoformat(s)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.timestamp()
except (ValueError, TypeError):
return 0.0
def _extract_outcome_price(meta: dict, outcome: str) -> Optional[float]:
"""Pull the current price for `outcome` from a /markets response.
/markets returns:
outcomes: ['Yes', 'No'] (JSON-encoded string or list)
outcomePrices: ['0.45', '0.55'] (same length, string floats)
We match outcome name case-insensitively and return the float price.
"""
outcomes_raw = meta.get("outcomes")
prices_raw = meta.get("outcomePrices")
if outcomes_raw is None or prices_raw is None:
return None
# /markets sometimes returns JSON-encoded strings for these fields
if isinstance(outcomes_raw, str):
try:
outcomes_raw = json.loads(outcomes_raw)
except (ValueError, TypeError):
return None
if isinstance(prices_raw, str):
try:
prices_raw = json.loads(prices_raw)
except (ValueError, TypeError):
return None
if not isinstance(outcomes_raw, list) or not isinstance(prices_raw, list):
return None
if len(outcomes_raw) != len(prices_raw) or not outcomes_raw:
return None
target = (outcome or "").strip().lower()
for i, name in enumerate(outcomes_raw):
if str(name).strip().lower() == target:
try:
p = float(prices_raw[i])
except (ValueError, TypeError):
return None
# Clamp to valid Polymarket price range
return max(0.0, min(1.0, p))
return None
@dataclass
class MarketAccumulator:
condition_id: str
market_question: str = ""
market_slug: str = ""
outcome: str = "Yes"
buy_strength: float = 0.0
sell_strength: float = 0.0
@@ -43,6 +130,37 @@ class MarketAccumulator:
last_trade_at: float = 0.0
window_started_at: float = 0.0
def to_dict(self) -> dict:
return {
"condition_id": self.condition_id,
"market_question": self.market_question,
"market_slug": self.market_slug,
"outcome": self.outcome,
"buy_strength": self.buy_strength,
"sell_strength": self.sell_strength,
"contributors_buy": self.contributors_buy,
"contributors_sell": self.contributors_sell,
"last_price": self.last_price,
"last_trade_at": self.last_trade_at,
"window_started_at": self.window_started_at,
}
@classmethod
def from_dict(cls, d: dict) -> "MarketAccumulator":
return cls(
condition_id=d["condition_id"],
market_question=d.get("market_question", ""),
market_slug=d.get("market_slug", ""),
outcome=d.get("outcome", "Yes"),
buy_strength=d.get("buy_strength", 0.0),
sell_strength=d.get("sell_strength", 0.0),
contributors_buy=d.get("contributors_buy", []),
contributors_sell=d.get("contributors_sell", []),
last_price=d.get("last_price", 0.0),
last_trade_at=d.get("last_trade_at", 0.0),
window_started_at=d.get("window_started_at", 0.0),
)
class SignalAggregator:
"""Detects consensus signals from a stream of wallet trades."""
@@ -52,15 +170,109 @@ class SignalAggregator:
db: CopyTraderDatabase,
sizer: Optional[KellySizer] = None,
data: Optional[DataAPIClient] = None,
risk_manager=None,
):
self.db = db
self.settings = get_settings()
self.sizer = sizer or KellySizer()
self.data = data or DataAPIClient()
self.data = data or get_shared_data_client()
self.risk = risk_manager # optional, set by main.py
self.accumulators: Dict[str, MarketAccumulator] = {}
self.wallet_credibility: Dict[str, float] = {}
self.wallet_debounce: Dict[str, float] = {}
# Same-market signal cooldown: cid → timestamp of last emit.
# Prevents spamming signals on one market within cooldown window.
self.recently_emitted: Dict[str, float] = {}
self._signals_emitted = 0
self._dirty = False
self._last_persist_at: float = 0.0
# market metadata cache: condition_id → (question, slug, end_date_ts, is_closed).
# end_date_ts is 0.0 when unknown. Used for min-hours-to-resolution filter.
# is_closed=True when the market is resolved/inactive → skip trading.
self._market_meta_cache: Dict[str, Tuple[str, str, float, bool]] = {}
self._load_accumulators()
self._load_debounce()
def _load_accumulators(self) -> None:
"""Restore in-flight consensus accumulators from DB (survives restart)."""
raw = self.db.get_pool_state_json("open_accumulators")
if not raw or not isinstance(raw, dict):
return
loaded = 0
now = time.time()
for cid, payload in raw.items():
last_trade_at = float(payload.get("last_trade_at", 0))
if now - last_trade_at > self.settings.consensus_window_seconds:
continue
try:
self.accumulators[cid] = MarketAccumulator.from_dict(payload)
loaded += 1
except Exception as e:
logger.debug(f"[agg] skip corrupt accumulator {cid[:10]}: {e}")
if loaded:
logger.info(f"[agg] restored {loaded} open accumulators from DB")
def _persist_accumulators(self) -> None:
"""Persist current accumulators to DB.
Throttled: at most one write per ``_PERSIST_DEBOUNCE_SECONDS``. The
``_dirty`` flag stays set until a write actually happens, so the final
state is not lost. Call ``flush_accumulators()`` on shutdown to force
a final write.
"""
if not self._dirty:
return
now = time.time()
if now - self._last_persist_at < _PERSIST_DEBOUNCE_SECONDS:
return
try:
payload = {cid: acc.to_dict() for cid, acc in self.accumulators.items()}
self.db.set_pool_state_json("open_accumulators", payload)
self._dirty = False
self._last_persist_at = now
except Exception as e:
logger.debug(f"[agg] persist accumulators failed: {e}")
def flush_accumulators(self) -> None:
"""Force a final accumulator write (used on shutdown)."""
if not self._dirty:
return
try:
payload = {cid: acc.to_dict() for cid, acc in self.accumulators.items()}
self.db.set_pool_state_json("open_accumulators", payload)
self._dirty = False
self._last_persist_at = time.time()
except Exception as e:
logger.debug(f"[agg] flush accumulators failed: {e}")
def _load_debounce(self) -> None:
"""Restore wallet_debounce from DB; drop entries older than the window."""
raw = self.db.get_pool_state_json(_DEBOUNCE_STATE_KEY)
if not raw or not isinstance(raw, dict):
return
now = time.time()
window = self.settings.wallet_debounce_seconds
loaded = 0
for key, ts_val in raw.items():
try:
ts = float(ts_val)
except (TypeError, ValueError):
continue
if now - ts < window:
self.wallet_debounce[key] = ts
loaded += 1
if loaded:
logger.info(f"[agg] restored {loaded} debounce entries from DB")
def _save_debounce(self) -> None:
"""Persist current wallet_debounce snapshot."""
try:
self.db.set_pool_state_json(_DEBOUNCE_STATE_KEY, self.wallet_debounce)
except Exception as e:
logger.debug(f"[agg] persist debounce failed: {e}")
def _mark_dirty(self) -> None:
self._dirty = True
def load_credibilities(self) -> None:
"""Bulk-load credibility from DB into memory."""
@@ -70,30 +282,98 @@ class SignalAggregator:
f"[agg] loaded {len(self.wallet_credibility)} wallet credibilities"
)
def update_credibility(self, address: str, credibility: float) -> None:
"""Apply credibility update from Bayesian updater."""
self.wallet_credibility[address] = credibility
self.db.update_wallet_credibility(address, credibility)
def update_credibility(
self,
address: str,
credibility: float,
realized_pnl_window: Optional[float] = None,
n_closed: Optional[int] = None,
) -> None:
"""Apply credibility update from Bayesian updater.
def _lazy_market_meta(self, cid: str, default_outcome: str) -> tuple:
"""Synchronously fetch market question; degrade gracefully on failure."""
try:
meta = self.data.get_market(cid)
if not meta:
return "", default_outcome
q = meta.get("question", "")
outcomes = meta.get("outcomes", "")
if isinstance(outcomes, str):
try:
outcomes = json.loads(outcomes)
except Exception:
outcomes = []
if isinstance(outcomes, list) and outcomes:
return q, outcomes[0]
return q, default_outcome
except Exception as e:
logger.debug(f"[agg] market meta fetch failed for {cid[:10]}: {e}")
return "", default_outcome
If realized_pnl_window / n_closed are provided they are recorded in
credibility_history for later auditing of the Bayesian formula.
"""
self.wallet_credibility[address] = credibility
if realized_pnl_window is not None or n_closed is not None:
self.db.update_wallet_credibility_with_context(
address, credibility, realized_pnl_window, n_closed,
)
else:
self.db.update_wallet_credibility(address, credibility)
def purge_wallet(self, address: str) -> None:
"""Remove all state for a wallet that's no longer in the pool."""
changed = False
if address in self.wallet_credibility:
self.wallet_credibility.pop(address, None)
changed = True
if address in self.wallet_debounce:
self.wallet_debounce.pop(address, None)
changed = True
for acc in self.accumulators.values():
before = len(acc.contributors_buy) + len(acc.contributors_sell)
acc.contributors_buy = [c for c in acc.contributors_buy if c["address"] != address]
acc.contributors_sell = [c for c in acc.contributors_sell if c["address"] != address]
if len(acc.contributors_buy) + len(acc.contributors_sell) != before:
changed = True
if changed:
self._dirty = True
logger.debug(f"[agg] purged wallet {address[:10]} from in-memory state")
async def _lazy_market_meta(self, cid: str) -> Tuple[str, str, float, bool]:
"""Fetch market question/slug/endDate/closed; degrade gracefully.
Returns (question, slug, end_date_ts, is_closed). end_date_ts is
0.0 when unknown. is_closed is True when the market is resolved or
inactive (closed=true or active=false in /markets response) — such
markets must not receive new signals. Cached per condition_id so
repeated trades on the same market don't re-hit Gamma API.
"""
cached = self._market_meta_cache.get(cid)
if cached is not None:
return cached
def _fetch() -> Tuple[str, str, float, bool]:
try:
meta = self.data.get_market(cid)
if not meta:
return "", "", 0.0, False
q = meta.get("question", "")
slug = meta.get("slug", "")
end_ts = _parse_end_date(meta.get("endDate"))
# /markets: closed (bool), active (bool). A market is
# tradeable only when active=true AND closed=false.
is_closed = bool(meta.get("closed")) or not bool(meta.get("active", True))
return q, slug, end_ts, is_closed
except Exception as e:
logger.debug(f"[agg] market meta fetch failed for {cid[:10]}: {e}")
return "", "", 0.0, False
result = await asyncio.to_thread(_fetch)
self._market_meta_cache[cid] = result
return result
async def _fetch_realtime_price(self, cid: str, outcome: str) -> Optional[float]:
"""Fetch current price for `outcome` from /markets (outcomePrices).
Returns None on failure so caller can fall back to last trade price.
Uses the meta cache when possible (already has endDate but not
prices), otherwise hits Gamma /markets directly.
"""
def _fetch() -> Optional[float]:
try:
meta = self.data.get_market(cid)
if not meta:
return None
return _extract_outcome_price(meta, outcome)
except Exception as e:
logger.debug(f"[agg] realtime price fetch failed for {cid[:10]}: {e}")
return None
return await asyncio.to_thread(_fetch)
@staticmethod
def _acc_key(cid: str, outcome: str) -> str:
"""Composite key: condition_id + outcome, so multi-outcome markets don't mix."""
return f"{cid}|{outcome}"
async def on_trade(self, wallet_addr: str, trade: dict) -> Optional[dict]:
"""Process a new trade. Returns a CopySignal dict if consensus reached."""
@@ -136,42 +416,104 @@ class SignalAggregator:
)
return None
# Debounce per (wallet, market)
deb_key = f"{wallet_addr}:{cid}"
last = self.wallet_debounce.get(deb_key, 0)
if time.time() - last < self.settings.wallet_debounce_seconds:
logger.debug(f"[agg] debounced {deb_key}")
# P0-3: BUY-only signals. SELL on Polymarket = selling tokens you
# already hold (not shorting). Without position tracking we cannot
# follow SELLs, so we log them as a reverse-direction indicator but
# do NOT trigger emission.
if side == "SELL" and self.settings.buy_only_signals:
logger.info(
f"[agg] SELL logged (no signal): {wallet_addr[:10]} ${usdc:.0f} "
f"@ {price:.4f} on {cid[:10]}/{outcome}"
)
return None
if side not in ("BUY", "SELL"):
return None
self.wallet_debounce[deb_key] = time.time()
acc = self.accumulators.get(cid)
# P1-7: same-market cooldown. Once a signal has fired for this
# condition_id, ignore all further trades on it until cooldown
# expires. Prevents spamming signals on the same market.
last_emit = self.recently_emitted.get(cid, 0)
if time.time() - last_emit < self.settings.signal_cooldown_seconds:
logger.debug(
f"[agg] {cid[:10]} in signal cooldown "
f"({int(self.settings.signal_cooldown_seconds - (time.time() - last_emit))}s left), skipping"
)
return None
# P2-10 + 附加: fetch market meta early. We need endDate for the
# min-hours-to-resolution filter, and closed/active flags to skip
# resolved markets. Doing this before accumulator creation ensures
# we never open an accumulator on a dead market.
market_q, market_slug, end_ts, is_closed = await self._lazy_market_meta(cid)
if is_closed:
logger.debug(f"[agg] {cid[:10]} is closed/inactive, skipping")
return None
if end_ts > 0:
hours_left = (end_ts - time.time()) / 3600.0
if hours_left < self.settings.min_hours_to_resolution:
logger.debug(
f"[agg] {cid[:10]} resolves in {hours_left:.1f}h "
f"< min {self.settings.min_hours_to_resolution}h, skipping"
)
return None
# Debounce per (wallet, market) — but only blocks NEW contributor
# registration. Add-position (bumping an existing contributor's
# size_usd) is allowed regardless of debounce so we don't lose
# follow-on size information (P2-8).
acc_key = self._acc_key(cid, outcome)
acc = self.accumulators.get(acc_key)
if acc is None:
market_q, derived_outcome = self._lazy_market_meta(cid, outcome)
acc = MarketAccumulator(
condition_id=cid,
market_question=market_q,
outcome=derived_outcome,
market_slug=market_slug,
outcome=outcome,
window_started_at=time.time(),
last_trade_at=time.time(),
)
self.accumulators[cid] = acc
logger.debug(f"[agg] new accumulator for {cid[:10]}: {market_q[:60]}")
self.accumulators[acc_key] = acc
self._mark_dirty()
logger.debug(f"[agg] new accumulator for {cid[:10]}/{outcome}: {market_q[:60]}")
acc.last_price = price
acc.last_trade_at = time.time()
contrib = {
"address": wallet_addr,
"credibility": cred,
"size_usd": usdc,
}
if side == "BUY":
acc.buy_strength += cred
acc.contributors_buy.append(contrib)
elif side == "SELL":
acc.sell_strength += cred
acc.contributors_sell.append(contrib)
contributors = acc.contributors_buy if side == "BUY" else acc.contributors_sell
existing = next((c for c in contributors if c["address"] == wallet_addr), None)
if existing is not None:
# P2-8: add-position — bump size only, do NOT re-add credibility
# strength (prevents one wallet inflating consensus by trading
# many small lots). size_usd accumulates so the signal still
# reflects the wallet's total conviction.
existing["size_usd"] += usdc
self._mark_dirty()
logger.debug(
f"[agg] {wallet_addr[:10]} add-position {side} +${usdc:.0f} "
f"(total ${existing['size_usd']:.0f}) on {cid[:10]}"
)
else:
return None
# New contributor — apply debounce to prevent re-registration
# within the debounce window.
deb_key = f"{wallet_addr}:{cid}"
last = self.wallet_debounce.get(deb_key, 0)
if time.time() - last < self.settings.wallet_debounce_seconds:
logger.debug(f"[agg] debounced {deb_key}")
return None
self.wallet_debounce[deb_key] = time.time()
contrib = {
"address": wallet_addr,
"credibility": cred,
"size_usd": usdc,
}
if side == "BUY":
acc.buy_strength += cred
acc.contributors_buy.append(contrib)
else:
acc.sell_strength += cred
acc.contributors_sell.append(contrib)
self._mark_dirty()
logger.debug(
f"[agg] [{cid[:10]}] {wallet_addr[:10]} {side} ${usdc:.0f} @ {price:.4f} | "
@@ -182,33 +524,74 @@ class SignalAggregator:
self._cleanup_expired()
await asyncio.to_thread(self._persist_accumulators)
spread = acc.buy_strength - acc.sell_strength
if (
len(acc.contributors_buy) >= self.settings.consensus_min_wallets
and spread >= self.settings.consensus_strength_threshold
):
return self._emit(cid, "BUY", acc.outcome, price, acc)
return await self._emit(cid, "BUY", outcome, price, acc, acc_key)
elif (
len(acc.contributors_sell) >= self.settings.consensus_min_wallets
and -spread >= self.settings.consensus_strength_threshold
):
return self._emit(cid, "SELL", acc.outcome, price, acc)
# SELL signals only fire when buy_only_signals=False. When True,
# SELL trades are blocked above and buy_strength/sell_strength
# stays symmetric (SELL branch unreachable in practice).
return await self._emit(cid, "SELL", outcome, price, acc, acc_key)
return None
def _emit(
async def _emit(
self,
cid: str,
side: str,
outcome: str,
price: float,
acc: MarketAccumulator,
) -> dict:
acc_key: str = "",
) -> Optional[dict]:
"""Build and emit a CopySignal. Returns None if risk gate blocks."""
contributors = acc.contributors_buy if side == "BUY" else acc.contributors_sell
n = len(contributors)
aggregated = sum(c["credibility"] for c in contributors) / max(1, n)
# P0-4: sum(credibility) * log1p(n). Rewards both count AND
# credibility with diminishing returns on count, so a single high-
# cred wallet can still trigger but a broad consensus is stronger.
# Old formula (mean) capped strength at 1.0 and made n irrelevant;
# this makes the threshold scale naturally with conviction.
aggregated = sum(c["credibility"] for c in contributors) * math.log1p(n)
total_size_usd = sum(c["size_usd"] for c in contributors)
kelly = self.sizer.fraction(aggregated, price, side)
# P1-6: risk-gate before doing any expensive work. If the circuit
# breaker is tripped (daily loss / consecutive losses / max open
# positions), skip emission entirely.
if self.risk is not None:
allowed, reason = await asyncio.to_thread(self.risk.can_emit)
if not allowed:
logger.warning(
f"[agg] signal blocked by risk gate: {reason} "
f"(cid={cid[:10]} side={side} strength={aggregated:.2f})"
)
return None
# P0-1: fetch real-time price from /markets as entry_price. The
# `price` passed in is the last trade price, which may be stale by
# up to USER_POLL_INTERVAL_SECONDS. Using a stale price for sizing
# means we under/over-size the position vs current liquidity.
rt_price = await self._fetch_realtime_price(cid, outcome)
if rt_price is not None and rt_price > 0:
entry_price = rt_price
price_source = "realtime"
else:
entry_price = price
price_source = "last_trade(fallback)"
if rt_price is not None and abs(rt_price - price) > 0.05:
logger.info(
f"[agg] price drift on {cid[:10]}: last_trade={price:.4f} "
f"realtime={rt_price:.4f} (using {rt_price:.4f})"
)
kelly = self.sizer.fraction(aggregated, entry_price, side)
suggested = self.sizer.position_usd(
kelly, self.settings.initial_capital_usd
)
@@ -220,9 +603,10 @@ class SignalAggregator:
signal = {
"condition_id": cid,
"market_question": acc.market_question,
"market_slug": acc.market_slug,
"side": side,
"outcome": outcome,
"entry_price": price,
"entry_price": entry_price,
"aggregated_strength": aggregated,
"source_wallets": wallet_contribs,
"kelly_fraction": kelly,
@@ -231,13 +615,21 @@ class SignalAggregator:
"n_contributors": n,
}
self._signals_emitted += 1
# P1-7: record cooldown so we don't re-emit on this market.
self.recently_emitted[cid] = time.time()
# P1-6: invalidate risk caches so the next can_emit() sees the new
# open-position count.
if self.risk is not None:
await asyncio.to_thread(self.risk.record_signal_emitted)
logger.info(
f"[agg] SIGNAL #{self._signals_emitted}: "
f"{side} {outcome} on {acc.market_question[:60]} @ {price:.4f} | "
f"{side} {outcome} on {acc.market_question[:60]} @ {entry_price:.4f} [{price_source}] | "
f"strength={aggregated:.2f} wallets={n} "
f"kelly={kelly:.3f} size=${suggested:.0f}"
)
self.accumulators.pop(cid, None)
self.accumulators.pop(acc_key or cid, None)
self._mark_dirty()
await asyncio.to_thread(self._persist_accumulators)
return signal
def _cleanup_expired(self) -> None:
@@ -250,6 +642,7 @@ class SignalAggregator:
for cid in expired:
acc = self.accumulators.pop(cid, None)
if acc:
self._mark_dirty()
logger.debug(
f"[agg] window expired for {cid[:10]} (no consensus reached)"
)
)
src/services/bayesian.py
+44 -34
View File
@@ -1,33 +1,35 @@
"""Bayesian credibility updater.
For each wallet, fetch recent closed-positions (realized PnL) and adjust
credibility using a simple posterior update.
credibility using a simple posterior update. Only positions within
BAYESIAN_DECAY_DAYS are considered (default 14 days).
State:
prior_skill: BAYESIAN_PRIOR_SKILL (default 0.5)
posterior: updated based on realized PnL trend
posterior: updated based on time-filtered realized PnL trend
Update rule (simple exponential smoothing):
if realized_30d > 0: cred += step * (1 - cred)
if realized_30d < 0: cred -= step * cred
if realized_recent > 0: cred += step * (1 - cred)
if realized_recent < 0: cred -= step * cred
Where step = 0.05 by default (slow update).
Concurrency: wallets are processed in parallel via a bounded semaphore.
"""
import asyncio
import logging
import time
from typing import Dict
from src.config import get_settings
from src.db.database import CopyTraderDatabase
from src.services.aggregator import SignalAggregator
from src.services.data_api import DataAPIClient
from src.services.data_api import DataAPIClient, get_shared_data_client
logger = logging.getLogger(__name__)
class BayesianUpdater:
"""Periodically refresh wallet credibility from realized PnL."""
"""Refresh wallet credibility from realized PnL."""
def __init__(
self,
@@ -38,34 +40,37 @@ class BayesianUpdater:
self.db = db
self.aggregator = aggregator
self.settings = get_settings()
self.data = data or DataAPIClient()
self._running = False
async def run(self) -> None:
self._running = True
interval = self.settings.credibility_update_minutes * 60
logger.info(f"[bayes] starting, interval={interval}s")
try:
while self._running:
await self._update_once()
await asyncio.sleep(interval)
except asyncio.CancelledError:
logger.info("[bayes] cancelled")
self.data = data or get_shared_data_client()
async def _update_once(self) -> None:
addresses = self.db.get_wallet_addresses()
if not addresses:
return
logger.debug(f"[bayes] updating credibility for {len(addresses)} wallets")
for addr in addresses:
try:
closed = await asyncio.to_thread(self.data.get_closed_positions, addr, 100)
except Exception as e:
logger.warning(f"[bayes] closed-positions fetch failed {addr[:10]}: {e}")
continue
sem = asyncio.Semaphore(self.settings.bayes_concurrency)
cutoff = int(time.time()) - self.settings.bayesian_decay_days * 86400
async def _one(addr: str) -> None:
async with sem:
try:
# get_closed_positions_since already filters by cutoff_ts,
# so all returned items are within the decay window.
# Use 3x the pool-building max_pages since bayesian only
# processes ~100 pool wallets and PnL accuracy directly
# affects credibility scores.
bayes_max_pages = self.settings.closed_positions_max_pages * 3
closed = await asyncio.to_thread(
self.data.get_closed_positions_since, addr, cutoff,
bayes_max_pages,
)
except Exception as e:
logger.warning(
f"[bayes] closed-positions fetch failed {addr[:10]}: {e}"
)
return
realized = sum(float(p.get("realizedPnl") or 0) for p in closed)
current = self.aggregator.wallet_credibility.get(addr, 0.5)
# Update
step = 0.05
step = self.settings.bayesian_step
if realized > 0:
new = current + step * (1 - current)
elif realized < 0:
@@ -73,11 +78,16 @@ class BayesianUpdater:
else:
new = current
new = max(0.05, min(0.95, new))
self.aggregator.update_credibility(addr, new)
self.aggregator.update_credibility(
addr, new,
realized_pnl_window=realized,
n_closed=len(closed),
)
logger.debug(
f"[bayes] {addr[:10]} realized=${realized:.0f} cred {current:.3f}→{new:.3f}"
f"[bayes] {addr[:10]} realized_recent=${realized:.0f} "
f"({len(closed)} pos in window) "
f"cred {current:.3f}→{new:.3f}"
)
logger.info(f"[bayes] credibility refresh complete")
def stop(self) -> None:
self._running = False
await asyncio.gather(*[_one(a) for a in addresses])
logger.info(f"[bayes] credibility refresh complete ({len(addresses)} wallets)")
src/services/data_api.py
+226 -8
View File
@@ -1,5 +1,6 @@
"""Polymarket public API clients (Gamma + Data). All endpoints are public, no auth."""
import logging
import threading
from typing import Any, Dict, List, Optional
from src.utils.http import get_client
@@ -11,7 +12,13 @@ DATA_BASE = "https://data-api.polymarket.com"
class DataAPIClient:
"""Client for Polymarket Data API (positions, holders, activity)."""
"""Client for Polymarket Data API (positions, holders, activity).
Use ``get_shared_data_client()`` instead of constructing directly when
possible — multiple services (wallet_pool, bayes, user_stream, aggregator)
each calling ``DataAPIClient()`` would otherwise spin up independent
httpx connection pools against the same host.
"""
def __init__(self):
self._client = get_client(timeout=30.0)
@@ -52,15 +59,168 @@ class DataAPIClient:
r.raise_for_status()
return r.json()
def get_closed_positions(self, address: str, limit: int = 100) -> List[dict]:
"""Closed positions (realized PnL)."""
def get_closed_positions(
self,
address: str,
limit: int = 50,
offset: int = 0,
sort_by: str = "TIMESTAMP",
sort_direction: str = "DESC",
timeout: Optional[float] = None,
) -> List[dict]:
"""Fetch a single page of closed positions (realized PnL).
API contract (docs/api-reference/core/get-closed-positions-for-a-user.md):
- limit: maximum 50 (server enforces; we clamp defensively)
- sortBy default REALIZEDPNL → returns most-profitable first, which
hides recent-but-small positions. Default to TIMESTAMP so callers
can filter by time correctly.
"""
kwargs: Dict[str, Any] = {}
if timeout is not None:
kwargs["timeout"] = timeout
r = self._client.get(
f"{DATA_BASE}/closed-positions",
params={"user": address, "limit": limit, "sortBy": "REALIZEDPNL"},
params={
"user": address,
"limit": max(0, min(limit, 50)),
"offset": max(0, offset),
"sortBy": sort_by,
"sortDirection": sort_direction,
},
**kwargs,
)
r.raise_for_status()
return r.json()
def get_closed_positions_since(
self,
address: str,
cutoff_ts: int,
max_pages: int = 10,
page_size: int = 50,
timeout: Optional[float] = None,
) -> List[dict]:
"""Fetch all closed positions newer than cutoff_ts, auto-paginating.
Uses sortBy=TIMESTAMP + DESC, so we can stop as soon as a record older
than cutoff_ts is encountered. Returns at most page_size * max_pages
records.
Pagination strategy: fetch page 0 first. If it's a full page (meaning
more data may exist), fetch pages 1..max_pages-1 CONCURRENTLY using a
thread pool, then merge + filter by cutoff_ts. This reduces the
worst-case latency from max_pages serial requests to 2 rounds
(1 + 1 parallel batch).
"""
import concurrent.futures
# Page 0: always fetch first to check if pagination is needed
first_page = self.get_closed_positions(
address,
limit=page_size,
offset=0,
sort_by="TIMESTAMP",
sort_direction="DESC",
timeout=timeout,
)
if not first_page:
return []
# Check if first page has records older than cutoff
collected: List[dict] = []
cutoff_reached = False
for item in first_page:
ts = item.get("timestamp")
if isinstance(ts, (int, float)) and ts < cutoff_ts:
cutoff_reached = True
break
collected.append(item)
# If first page is partial or cutoff reached, no need for more pages
if len(first_page) < page_size or cutoff_reached:
return collected
# Full first page + no cutoff hit → fetch remaining pages concurrently
remaining_pages = max_pages - 1
if remaining_pages <= 0:
logger.warning(
f"[data_api] get_closed_positions_since hit max_pages={max_pages} "
f"cap for {address[:10]} ({len(collected)} items since cutoff); "
f"older records in the window are NOT included"
)
return collected
offsets = [(p + 1) * page_size for p in range(remaining_pages)]
def _fetch_page(off: int) -> List[dict]:
return self.get_closed_positions(
address,
limit=page_size,
offset=off,
sort_by="TIMESTAMP",
sort_direction="DESC",
timeout=timeout,
)
# Use a thread pool to fetch pages concurrently. max_workers is
# capped to avoid overwhelming the API (5 concurrent pages).
# Overall timeout: cap the whole parallel batch to 15s so a single
# slow/hanging page doesn't stall the wallet-pool pipeline (we'd
# rather under-count PnL on one wallet than block the whole build).
max_workers = min(remaining_pages, 5)
page_results: List[List[dict]] = [[] for _ in range(remaining_pages)]
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as pool:
future_to_idx = {
pool.submit(_fetch_page, off): idx
for idx, off in enumerate(offsets)
}
try:
for future in concurrent.futures.as_completed(
future_to_idx, timeout=15.0,
):
idx = future_to_idx[future]
try:
page_results[idx] = future.result()
except Exception:
page_results[idx] = []
except concurrent.futures.TimeoutError:
# Some pages didn't finish in 15s — cancel the rest and
# proceed with whatever we have. Partial PnL is acceptable
# for the pool-build phase (bayes will refine later).
not_done = [f for f in future_to_idx if not f.done()]
logger.debug(
f"[data_api] closed_positions parallel batch timed out "
f"for {address[:10]}: {len(not_done)}/{len(offsets)} pages "
f"unfinished, proceeding with partial data"
)
for f in not_done:
f.cancel()
# Merge pages in order, stop at cutoff
hit_cap = True
for page_items in page_results:
if not page_items:
hit_cap = False
break
for item in page_items:
ts = item.get("timestamp")
if isinstance(ts, (int, float)) and ts < cutoff_ts:
hit_cap = False
break
collected.append(item)
if len(page_items) < page_size:
hit_cap = False
break
if hit_cap and len(collected) >= page_size * max_pages:
logger.warning(
f"[data_api] get_closed_positions_since hit max_pages={max_pages} "
f"cap for {address[:10]} ({len(collected)} items since cutoff); "
f"older records in the window are NOT included"
)
return collected
def get_activity(self, address: str, limit: int = 50) -> List[dict]:
"""Fetch recent activity (TRADE, SPLIT, MERGE, etc.) for a wallet."""
r = self._client.get(
@@ -76,15 +236,47 @@ class DataAPIClient:
r.raise_for_status()
return r.json()
def get_trades(
self,
address: str,
limit: int = 1000,
start_ts: Optional[int] = None,
end_ts: Optional[int] = None,
) -> List[dict]:
"""Fetch trades for a wallet with optional time window.
Response fields: transactionHash, side, size, price, timestamp,
conditionId, eventSlug, slug, outcome, outcomeIndex, etc.
"""
params: Dict[str, Any] = {
"user": address,
"limit": min(limit, 10000),
"takerOnly": "true",
}
if start_ts is not None:
params["start"] = start_ts
if end_ts is not None:
params["end"] = end_ts
r = self._client.get(
f"{DATA_BASE}/trades",
params=params,
)
r.raise_for_status()
return r.json()
def get_market(self, condition_id: str) -> Optional[dict]:
"""Single market metadata from Gamma API."""
"""Single market metadata from Gamma API. Returns {'question','slug','outcomes',...} or None."""
try:
r = self._client.get(
f"{GAMMA_BASE}/markets/{condition_id}",
f"{GAMMA_BASE}/markets",
params={"condition_id": condition_id},
timeout=10,
)
r.raise_for_status()
return r.json()
except Exception:
data = r.json()
return data[0] if data else None
except Exception as e:
logger.warning(f"get_market failed for {condition_id[:16]}: {e}")
return None
@@ -115,3 +307,29 @@ class GammaAPIClient:
r.raise_for_status()
data = r.json()
return data[0] if data else None
# ----- Shared singletons -----
# Single process-wide DataAPIClient + GammaAPIClient so all services share
# one httpx connection pool per host instead of N independent pools.
_shared_data_client: Optional["DataAPIClient"] = None
_shared_gamma_client: Optional["GammaAPIClient"] = None
_shared_lock = threading.Lock()
def get_shared_data_client() -> "DataAPIClient":
global _shared_data_client
if _shared_data_client is None:
with _shared_lock:
if _shared_data_client is None:
_shared_data_client = DataAPIClient()
return _shared_data_client
def get_shared_gamma_client() -> "GammaAPIClient":
global _shared_gamma_client
if _shared_gamma_client is None:
with _shared_lock:
if _shared_gamma_client is None:
_shared_gamma_client = GammaAPIClient()
return _shared_gamma_client
src/services/kelly.py
+109 -12
View File
@@ -1,8 +1,20 @@
"""Kelly position sizer with Favorite-Longshot bias correction.
"""Kelly position sizer with empirical win-probability estimation.
Two modes:
1. **Fallback** (default, when < KELLY_MIN_SAMPLES resolved signals exist):
Use a conservative fixed fraction (KELLY_FALLBACK_FRACTION, default 1%).
The old "p = 0.50 + 0.35*strength" formula was a placeholder with no
empirical basis — using it for real money is gambling.
2. **Empirical** (once enough history is collected):
Query the DB for the actual hit-rate of signals at similar strength
levels, and use that as p in the Kelly formula. This makes Kelly
self-calibrating: as the bot runs longer, p converges to the true
win-probability.
Kelly formula (half-Kelly by default for safety):
p = estimated win probability (from signal strength)
b = payoff ratio: (1 - price)/price for BUY, price/(1-price) for SELL
p = estimated win probability (from empirical hit-rate)
b = payoff ratio: (1 - price)/price for BUY
q = 1 - p
f* = (b * p - q) / b
use = f* * kelly_fraction (default 0.5 → half-Kelly)
@@ -11,26 +23,95 @@ Favorite-Longshot bias correction (research-driven):
factor = (1 - 2 * |price - 0.5|)^beta
Edge at extreme prices (<0.10 or >0.90) is reduced.
"""
import logging
import math
from typing import Optional
from src.config import get_settings
logger = logging.getLogger(__name__)
def _parse_bucket_label(label: str) -> tuple:
"""Parse a strength bucket label into (lo, hi) floats.
Supports two formats:
'0.5-1.0' → (0.5, 1.0)
'5.0+' → (5.0, float('inf'))
Returns (None, None) on parse failure.
"""
if not label or not isinstance(label, str):
return None, None
label = label.strip()
if label.endswith("+"):
try:
lo = float(label[:-1])
return lo, float("inf")
except ValueError:
return None, None
parts = label.split("-")
if len(parts) != 2:
return None, None
try:
return float(parts[0]), float(parts[1])
except ValueError:
return None, None
class KellySizer:
"""Position sizing per the research framework."""
def __init__(self):
def __init__(self, db=None):
self.settings = get_settings()
self._db = db
# Cache empirical hit-rate for 5 minutes to avoid hammering the DB
self._hr_cache: Optional[dict] = None
self._hr_cache_ts: float = 0
self._hr_cache_ttl = 300.0
def win_probability(self, strength: float) -> float:
"""Map signal strength [0,1] → win probability.
def _empirical_hit_rate(self, strength: float) -> Optional[float]:
"""Look up actual hit-rate for signals at this strength level.
strength=0.5 → p=0.55 (baseline)
strength=1.0 → p=0.85 (strong consensus)
strength=0.0 → p=0.50 (coin flip)
Returns None if insufficient data (< kelly_min_samples resolved).
"""
strength = max(0.0, min(1.0, strength))
return 0.50 + 0.35 * strength
import time
now = time.time()
if self._hr_cache is not None and now - self._hr_cache_ts < self._hr_cache_ttl:
buckets = self._hr_cache
else:
if self._db is None:
return None
try:
buckets = self._db.get_signal_stats_by_strength()
except Exception:
return None
self._hr_cache = buckets
self._hr_cache_ts = now
# Find the bucket containing this strength
total_n = sum(b.get("n", 0) for b in buckets)
if total_n < self.settings.kelly_min_samples:
return None
# Buckets are labeled like "0.5-1.0" or "5.0+" (open-ended).
# Parse each and find the one whose range contains `strength`.
for b in buckets:
label = b.get("bucket", "")
n = b.get("n", 0)
wins = b.get("wins", 0)
lo, hi = _parse_bucket_label(label)
if lo is None:
continue
if lo <= strength < hi and n > 0:
return wins / n
# Strength didn't fit any bucket — use overall hit-rate
total_wins = sum(b.get("wins", 0) for b in buckets)
return total_wins / total_n if total_n > 0 else None
def win_probability(self, strength: float) -> Optional[float]:
"""Return empirical win-probability for this strength, or None if
insufficient data (caller should use fallback)."""
return self._empirical_hit_rate(strength)
def payoff_ratio(self, price: float, side: str) -> float:
"""How much we win vs how much we risk."""
@@ -51,8 +132,24 @@ class KellySizer:
side: str,
beta: float = 1.5,
) -> float:
"""Compute Kelly fraction (capped 0..1) for a single signal."""
"""Compute Kelly fraction (capped 0..1) for a single signal.
Uses empirical hit-rate as p when enough history exists; otherwise
falls back to a conservative fixed fraction.
"""
p = self.win_probability(signal_strength)
if p is None:
# Not enough historical data → conservative fixed fraction
logger.debug(
f"[kelly] insufficient history, using fallback fraction "
f"{self.settings.kelly_fallback_fraction}"
)
return min(
self.settings.kelly_fallback_fraction,
self.settings.max_position_pct,
)
q = 1.0 - p
b = self.payoff_ratio(price, side)
src/services/outcome_resolver.py
+178
View File
@@ -0,0 +1,178 @@
"""Outcome resolver: backfill resolved-market results onto copy_signals.
For every signal where the underlying market has resolved, fetch the
winning outcome from the Gamma /markets API and write back:
- exit_price (1.0 if our outcome won, else 0.0)
- pnl_usd (theoretical PnL if the suggested_size had been taken)
- outcome_correct (1 / 0)
- resolved_at (market's closedTime)
This is the foundation of all strategy-optimization statistics: without it
we know which signals were emitted but never which ones made money.
PnL model
---------
A BUY signal at entry_price p with suggested_size S means we notionally
buy S/p shares. On resolution:
- win : each share pays $1 → PnL = (1 - p) * (S / p) = S * (1 - p) / p
- loss : each share pays $0 → PnL = -p * (S / p) = -S
This is the theoretical "fully filled at entry_price" PnL — real execution
slippage is not modelled here. The number is for strategy evaluation, not
accounting.
"""
import asyncio
import json
import logging
from typing import List, Optional
from src.config import get_settings
from src.db.database import CopyTraderDatabase
from src.services.data_api import DataAPIClient, get_shared_data_client
logger = logging.getLogger(__name__)
class OutcomeResolver:
"""Backfills resolved-market results onto copy_signals."""
def __init__(
self,
db: CopyTraderDatabase,
data: Optional[DataAPIClient] = None,
):
self.db = db
self.settings = get_settings()
self.data = data or get_shared_data_client()
async def resolve_once(self, max_signals: Optional[int] = None) -> dict:
"""Scan unresolved signals, fetch market resolutions, write back.
Returns a dict with counters: scanned / resolved / skipped / failed.
"""
batch_size = max_signals or self.settings.outcome_resolve_batch_size
pending = self.db.get_unresolved_signals(limit=batch_size)
if not pending:
return {"scanned": 0, "resolved": 0, "skipped": 0, "failed": 0}
sem = asyncio.Semaphore(self.settings.wallet_pool_concurrency)
timeout = self.settings.outcome_resolve_request_timeout
resolved = 0
skipped = 0
failed = 0
unique_cids = {p["condition_id"] for p in pending}
# Cache market resolution per condition_id to avoid refetching
# the same market for multiple signals on the same market.
cache: dict = {}
async def fetch_resolution(cid: str) -> Optional[dict]:
if cid in cache:
return cache[cid]
async with sem:
try:
meta = await asyncio.wait_for(
asyncio.to_thread(self.data.get_market, cid),
timeout=timeout,
)
except (asyncio.TimeoutError, Exception) as e:
logger.debug(f"[resolve] get_market failed {cid[:10]}: {e}")
cache[cid] = None
return None
if not meta:
cache[cid] = None
return None
resolution = self._extract_resolution(meta)
cache[cid] = resolution
return resolution
# Prefetch all unique markets in parallel
resolutions = await asyncio.gather(
*[fetch_resolution(cid) for cid in unique_cids]
)
cid_to_resolution = dict(zip(unique_cids, resolutions))
for sig in pending:
cid = sig["condition_id"]
resolution = cid_to_resolution.get(cid)
if resolution is None:
failed += 1
continue
if not resolution["closed"]:
# Market not yet resolved; skip silently
skipped += 1
continue
winning_outcome = resolution.get("winning_outcome")
if winning_outcome is None:
# Closed but no clear winner (e.g. cancelled). Mark as skipped
# so we don't refetch every cycle — set resolved_at so it
# leaves the pending queue.
self.db.mark_signal_resolved_unpnl(
sig["id"], resolution.get("closed_time", ""),
)
skipped += 1
continue
sig_outcome = sig["outcome"]
entry_price = float(sig["entry_price"])
size_usd = float(sig["suggested_size_usd"])
correct = 1 if sig_outcome == winning_outcome else 0
exit_price = 1.0 if correct else 0.0
if correct:
pnl = size_usd * (1 - entry_price) / entry_price if entry_price > 0 else 0.0
else:
pnl = -size_usd
self.db.update_signal_outcome(
signal_id=sig["id"],
exit_price=exit_price,
pnl_usd=pnl,
outcome_correct=correct,
resolved_at=resolution.get("closed_time", ""),
)
resolved += 1
logger.info(
f"[resolve] batch done: scanned={len(pending)} resolved={resolved} "
f"skipped={skipped} failed={failed}"
)
return {
"scanned": len(pending),
"resolved": resolved,
"skipped": skipped,
"failed": failed,
}
@staticmethod
def _extract_resolution(market: dict) -> dict:
"""Parse /markets response into a resolution record.
Per Gamma API docs:
- ``closed`` (bool): market is closed
- ``closedTime`` (str): closure timestamp
- ``outcomes`` (str): JSON array, e.g. '["Yes", "No"]'
- ``outcomePrices`` (str): JSON array of prices, e.g. '["1", "0"]'
On resolution the winning outcome's price is 1, loser's is 0.
"""
closed = bool(market.get("closed", False))
closed_time = market.get("closedTime") or ""
winning_outcome = None
if closed:
try:
outcomes = json.loads(market.get("outcomes") or "[]")
prices = json.loads(market.get("outcomePrices") or "[]")
for outcome, price in zip(outcomes, prices):
try:
p = float(price)
except (TypeError, ValueError):
continue
if p >= 0.99: # tolerate rounding
winning_outcome = outcome
break
except (json.JSONDecodeError, TypeError):
pass
return {
"closed": closed,
"closed_time": closed_time,
"winning_outcome": winning_outcome,
}
src/services/risk_manager.py
+136
View File
@@ -0,0 +1,136 @@
"""Risk manager: gate signal emission with circuit-breaker rules.
Checks before emitting any signal:
1. Daily loss cap — if today's realized PnL < -MAX_DAILY_LOSS_USD, pause
2. Consecutive loss streak — if last N signals all lost, pause
3. Max open positions — if N unresolved signals already in flight, skip
Pause state is in-memory (resets on restart). This is intentional: a restart
should be a conscious human action that resets the circuit breaker.
"""
import logging
import time
from typing import Optional
from src.config import get_settings
from src.db.database import CopyTraderDatabase
logger = logging.getLogger(__name__)
class RiskManager:
"""Circuit-breaker gating for signal emission."""
def __init__(self, db: CopyTraderDatabase):
self.db = db
self.settings = get_settings()
self._paused_until: float = 0.0
self._pause_reason: str = ""
# Cache daily stats for 60s to avoid DB hammering on every signal
self._daily_cache_ts: float = 0.0
self._daily_cache: dict = {}
self._streak_cache_ts: float = 0.0
self._streak_cache: int = 0
def _is_paused(self) -> bool:
if self._paused_until > time.time():
return True
if self._paused_until > 0 and self._paused_until <= time.time():
logger.info(f"[risk] pause expired ({self._pause_reason}), resuming")
self._paused_until = 0.0
self._pause_reason = ""
return False
def _pause(self, reason: str, minutes: int) -> None:
self._paused_until = time.time() + minutes * 60
self._pause_reason = reason
logger.warning(f"[risk] PAUSED for {minutes}min: {reason}")
def _today_pnl(self) -> float:
"""Sum of pnl_usd for signals resolved in the last 24h."""
now = time.time()
if now - self._daily_cache_ts < 60:
return self._daily_cache.get("pnl", 0.0)
try:
rows = self.db.get_recent_signals(limit=500)
cutoff = now - 86400
total = 0.0
for s in rows:
resolved = s.get("resolved_at")
if not resolved:
continue
pnl = float(s.get("pnl_usd") or 0)
total += pnl
self._daily_cache = {"pnl": total}
self._daily_cache_ts = now
return total
except Exception as e:
logger.debug(f"[risk] daily pnl calc failed: {e}")
return 0.0
def _consecutive_losses(self) -> int:
"""Count consecutive losing signals (most recent first)."""
now = time.time()
if now - self._streak_cache_ts < 60:
return self._streak_cache
try:
rows = self.db.get_recent_signals(limit=100)
streak = 0
for s in rows:
correct = s.get("outcome_correct")
if correct is None:
continue # unresolved, skip
if correct == 0:
streak += 1
else:
break # win or cancelled → streak ends
self._streak_cache = streak
self._streak_cache_ts = now
return streak
except Exception as e:
logger.debug(f"[risk] streak calc failed: {e}")
return 0
def _open_position_count(self) -> int:
"""Count unresolved BUY signals (proxy for open positions)."""
try:
rows = self.db.get_recent_signals(limit=500)
return sum(1 for s in rows if not s.get("resolved_at"))
except Exception:
return 0
def can_emit(self) -> tuple[bool, str]:
"""Check all risk gates. Returns (allowed, reason_if_blocked)."""
if self._is_paused():
remaining = int((self._paused_until - time.time()) / 60)
return False, f"paused ({self._pause_reason}, {remaining}min left)"
# Gate 1: daily loss cap
daily_pnl = self._today_pnl()
if daily_pnl < -self.settings.max_daily_loss_usd:
self._pause(
f"daily loss ${daily_pnl:.0f} < -${self.settings.max_daily_loss_usd:.0f}",
self.settings.risk_pause_minutes,
)
return False, self._pause_reason
# Gate 2: consecutive losses
streak = self._consecutive_losses()
if streak >= self.settings.max_consecutive_losses:
self._pause(
f"{streak} consecutive losses",
self.settings.risk_pause_minutes,
)
return False, self._pause_reason
# Gate 3: max open positions
open_count = self._open_position_count()
if open_count >= self.settings.max_open_positions:
return False, f"max open positions reached ({open_count})"
return True, ""
def record_signal_emitted(self) -> None:
"""Invalidate caches after a new signal is emitted."""
self._daily_cache_ts = 0.0
self._streak_cache_ts = 0.0
src/services/telegram.py
+33 -11
View File
@@ -24,31 +24,43 @@ class TelegramNotifier:
return
try:
from telegram import Bot
self._bot = Bot(token=self.settings.telegram_bot_token)
from telegram.request import HTTPXRequest
proxy = self.settings.http_proxy
kwargs = {}
if proxy:
kwargs["request"] = HTTPXRequest(proxy=proxy)
self._bot = Bot(token=self.settings.telegram_bot_token, **kwargs)
me = await self._bot.get_me()
logger.info(f"Telegram bot started: @{me.username}")
logger.info(f"Telegram bot started: @{me.username} (proxy={proxy})")
except Exception as e:
logger.error(f"Failed to start Telegram bot: {e}")
self._bot = None
async def send_signal(self, market: str, side: str, strength: float,
size_usd: float, n_wallets: int) -> None:
size_usd: float, n_wallets: int,
condition_id: str = "", outcome: str = "",
entry_price: float = 0.0,
market_slug: str = "") -> None:
if not self._bot:
return
ref = market_slug or condition_id
link = f"https://polymarket.com/market/{ref}" if ref else ""
text = (
f"🎯 *Copy Signal*\n"
f"Market: {market[:80]}\n"
f"Direction: {side}\n"
f"Strength: {strength:.2f}\n"
f"Suggested size: ${size_usd:,.0f}\n"
f"Source wallets: {n_wallets}\n"
f"🎯 *跟单信号*\n"
f"市场：{market[:80]}\n"
f"{'链接：' + link + '\n' if link else ''}"
f"方向：{side} {outcome}\n"
f"入场价：${entry_price:.4f}\n"
f"信号强度：{strength:.2f}\n"
f"建议仓位：${size_usd:,.0f}\n"
f"来源钱包：{n_wallets}\n"
)
try:
await self._bot.send_message(
chat_id=self.settings.telegram_chat_id,
text=text,
parse_mode="Markdown",
disable_web_page_preview=True,
disable_web_page_preview=False,
)
except Exception as e:
logger.error(f"Failed to send Telegram signal: {e}")
@@ -59,7 +71,17 @@ class TelegramNotifier:
try:
await self._bot.send_message(
chat_id=self.settings.telegram_chat_id,
text=f"⚠️ Error: {message[:500]}",
text=f"⚠️ 错误：{message[:500]}",
)
except Exception:
pass
async def close(self) -> None:
"""Clean up bot resources (call on shutdown)."""
if self._bot is not None:
try:
# python-telegram-bot Bot doesn't require explicit close, but
# we drop the reference to allow GC of underlying httpx client.
self._bot = None
except Exception:
pass
src/services/user_stream.py
+116 -14
View File
@@ -5,20 +5,37 @@ Strategy:
- New trade = trade whose tx_hash is not in last_seen set
- Skip during warmup (avoid historical-trade noise)
- Yield trade via callback for aggregator
Restart-safety
--------------
``_last_seen`` is persisted to the ``pool_state`` table so that a process
restart does NOT treat the entire recent trade history as "new". Without this,
every restart either (a) floods the aggregator with stale trades that look
new, or (b) drops real trades during the warmup that follows a fresh start.
Bug fixed vs. original implementation:
The old code only set ``_last_seen[address]`` when ``trades`` was non-empty.
If the first poll returned [] (transient API failure / new wallet with no
activity yet), ``_last_seen[address]`` stayed ``None``. The next poll would
then iterate the full returned list with ``tx_hash == None`` never matching
→ ALL historical trades treated as new → spurious signals. We now mark the
baseline (empty string) even on empty responses.
"""
import asyncio
import logging
import time
from typing import Awaitable, Callable, Dict, List, Optional, Set
from typing import Awaitable, Callable, Dict, List, Optional
from src.config import get_settings
from src.db.database import CopyTraderDatabase
from src.services.data_api import DataAPIClient
from src.services.data_api import DataAPIClient, get_shared_data_client
logger = logging.getLogger(__name__)
TradeCallback = Callable[[str, dict], Awaitable[None]]
_STATE_KEY = "stream_last_seen"
class UserTradeStream:
"""Polls each tracked wallet and emits new trades."""
@@ -26,20 +43,55 @@ class UserTradeStream:
def __init__(self, db: CopyTraderDatabase, data: Optional[DataAPIClient] = None):
self.db = db
self.settings = get_settings()
self.data = data or DataAPIClient()
self.data = data or get_shared_data_client()
# address → most-recent tx_hash seen (or "" if baseline-empty).
# ``None`` means "never polled yet".
self._last_seen: Dict[str, Optional[str]] = {}
self._failure_count: Dict[str, int] = {}
self._backoff_until: Dict[str, float] = {}
self._running = False
self._poll_count = 0
self._trades_emitted = 0
self._load_last_seen()
# ----- state persistence -----
def _load_last_seen(self) -> None:
"""Restore last_seen from DB so restarts don't replay history."""
raw = self.db.get_pool_state_json(_STATE_KEY)
if not raw or not isinstance(raw, dict):
return
loaded = 0
for addr, h in raw.items():
if isinstance(h, str):
self._last_seen[addr] = h
loaded += 1
if loaded:
logger.info(f"[stream] restored last_seen for {loaded} wallets from DB")
def _save_last_seen(self) -> None:
"""Persist current last_seen snapshot. Called periodically + on stop."""
# Only persist non-None entries; None = never polled.
payload = {a: h for a, h in self._last_seen.items() if h is not None}
try:
self.db.set_pool_state_json(_STATE_KEY, payload)
except Exception as e:
logger.debug(f"[stream] persist last_seen failed: {e}")
async def run(self, on_trade: TradeCallback) -> None:
"""Continuously poll wallets and call on_trade(wallet_addr, trade)."""
self._running = True
warmup_until = time.time() + self.settings.stream_warmup_seconds
# warmup must be ≥ 2× poll interval so the first non-empty poll can
# establish a baseline before real trades are admitted.
warmup = max(
self.settings.stream_warmup_seconds,
2 * self.settings.user_poll_interval_seconds,
)
warmup_until = time.time() + warmup
logger.info(
f"[stream] starting — poll_interval={self.settings.user_poll_interval_seconds}s, "
f"warmup={self.settings.stream_warmup_seconds}s"
f"warmup={warmup}s"
)
last_save = time.time()
try:
while self._running:
self._poll_count += 1
@@ -52,6 +104,14 @@ class UserTradeStream:
sem = asyncio.Semaphore(self.settings.wallet_pool_concurrency)
poll_sem = self.settings.wallet_pool_concurrency
poll_timeout = self.settings.wallet_pool_request_timeout
now_ts = time.time()
address_list = [
a for a in addresses
if self._backoff_until.get(a, 0) <= now_ts
]
skipped = len(addresses) - len(address_list)
if skipped and self._poll_count % 10 == 0:
logger.debug(f"[stream] skipped {skipped} wallets in backoff")
async def poll_one(addr: str) -> None:
async with sem:
@@ -63,25 +123,40 @@ class UserTradeStream:
),
timeout=poll_timeout,
)
self._failure_count.pop(addr, None)
self._backoff_until.pop(addr, None)
await self._process_trades(addr, trades, on_trade, in_warmup)
except asyncio.TimeoutError:
logger.debug(f"[stream] poll timeout for {addr[:10]}")
except Exception as e:
logger.warning(
f"[stream] poll failed for {addr[:10]}: {e}",
exc_info=self.settings.log_level == "DEBUG",
)
except (asyncio.TimeoutError, Exception) as e:
n_fail = self._failure_count.get(addr, 0) + 1
self._failure_count[addr] = n_fail
backoff = min(300, 2 ** min(n_fail, 8))
self._backoff_until[addr] = time.time() + backoff
if n_fail >= 3:
logger.warning(
f"[stream] poll failed for {addr[:10]} "
f"({n_fail}x, backoff {backoff}s): {type(e).__name__}"
)
await asyncio.gather(*[poll_one(addr) for addr in addresses])
if self._poll_count % 10 == 0:
if self._poll_count % 10 == 0 or self.settings.stream_verbose_logging:
logger.info(
f"[stream] poll #{self._poll_count} complete "
f"({len(addresses)} wallets, {self._trades_emitted} total trades emitted)"
)
# Persist last_seen every ~5 min (each poll is ~30s).
now = time.time()
if now - last_save > 300:
self._save_last_seen()
last_save = now
await asyncio.sleep(self.settings.user_poll_interval_seconds)
except asyncio.CancelledError:
logger.info("[stream] cancelled, shutting down")
finally:
# Always flush state on exit so next restart resumes correctly.
self._save_last_seen()
async def _process_trades(
self,
@@ -91,19 +166,29 @@ class UserTradeStream:
in_warmup: bool,
) -> None:
last_hash = self._last_seen.get(address)
never_polled = last_hash is None
new_trades: List[dict] = []
for trade in trades:
tx_hash = self._trade_hash(trade)
if tx_hash is None:
logger.debug(
f"[stream] {address[:10]} trade without hash key, skipping: {trade}"
)
continue
if tx_hash == last_hash:
break
new_trades.append(trade)
# CRITICAL: establish baseline even when trades is empty. Otherwise a
# later non-empty response would replay its entire list as "new".
# Empty string sentinel = "polled, no baseline hash yet".
if trades:
first_hash = self._trade_hash(trades[0])
if first_hash is not None:
self._last_seen[address] = first_hash
elif never_polled:
self._last_seen[address] = ""
if in_warmup:
logger.info(
@@ -124,7 +209,11 @@ class UserTradeStream:
@staticmethod
def _trade_hash(trade: dict) -> Optional[str]:
"""Stable unique key per Polymarket trade record."""
"""Stable unique key per Polymarket trade record.
Per docs: /activity response includes ``transactionHash``. We also
accept ``tx_hash``/``id``/``tradeId`` as defensive fallbacks.
"""
for k in ("transactionHash", "tx_hash", "id", "tradeId"):
v = trade.get(k)
if v:
@@ -133,3 +222,16 @@ class UserTradeStream:
def stop(self) -> None:
self._running = False
def prune_last_seen(self, active_addresses: List[str]) -> None:
"""Drop _last_seen / _failure_count / _backoff entries for addresses no longer in pool."""
active = set(active_addresses)
stale = set(self._last_seen) - active
for addr in stale:
self._last_seen.pop(addr, None)
self._failure_count.pop(addr, None)
self._backoff_until.pop(addr, None)
if stale and self.settings.stream_verbose_logging:
logger.debug(f"[stream] pruned {len(stale)} stale wallet entries")
if stale:
self._save_last_seen()
src/services/wallet_pool.py
+303 -95
View File
@@ -6,26 +6,54 @@ Strategy:
3. Deduplicate wallet addresses across all markets
4. For each candidate (parallel), fetch /positions to compute PnL/trades/categories
5. Apply health score; keep top N by score; persist to DB
Resume support
--------------
Phase 3 (per-wallet profiling) is the slowest part — 19000+ wallets at
~1s each. To survive restarts, the builder persists:
- candidates dict (Phase 2 output) under ``_pool_build_candidates``
- scanned wallet set + passed wallets under ``_pool_build_progress``
On restart, if a valid checkpoint exists (age < _RESUME_TTL_SECONDS), the
builder skips Phase 1+2 and resumes Phase 3 from where it left off.
The checkpoint is cleared on successful completion.
"""
import asyncio
import logging
from datetime import datetime
import time
from datetime import datetime, timezone
from typing import Dict, List, Set
from src.config import get_settings
from src.db.database import CopyTraderDatabase
from src.services.data_api import DataAPIClient, GammaAPIClient
from src.services.data_api import get_shared_data_client, get_shared_gamma_client
logger = logging.getLogger(__name__)
# Checkpoint keys in pool_state table
_CANDIDATES_KEY = "_pool_build_candidates"
_PROGRESS_KEY = "_pool_build_progress"
_RESUME_TTL_SECONDS = 2 * 3600 # discard checkpoint older than 2h
_PERSIST_EVERY = 50 # persist progress every N scanned wallets
def compute_health_score(pnl_30d: float, total_pnl: float,
trades: int, categories: int) -> float:
"""0-100 score combining PnL magnitude, trade count, category diversity."""
if trades < 10 or pnl_30d <= 0:
def compute_health_score(
pnl: float,
trades: int,
categories: int,
min_trades: int = 10,
) -> float:
"""0-100 score combining PnL magnitude, trade count, category diversity.
pnl: total PnL from current positions (cashPnl + realizedPnl)
trades: actual trade count from /trades endpoint (via transactionHash dedup)
categories: unique eventSlug count from /trades + /positions
min_trades: floor below which score is 0; pass settings.wallet_min_trades
so the score gate matches the pool filter gate.
"""
if trades < min_trades or pnl <= 0:
return 0.0
pnl_score = min(pnl_30d / 10000.0, 1.0) * 50
pnl_score = min(pnl / 10000.0, 1.0) * 50
trade_score = min(trades / 100.0, 1.0) * 25
diversity_score = min(categories / 5.0, 1.0) * 25
@@ -38,12 +66,81 @@ class WalletPoolBuilder:
def __init__(self, db: CopyTraderDatabase):
self.db = db
self.settings = get_settings()
self.gamma = GammaAPIClient()
self.data = DataAPIClient()
self.gamma = get_shared_gamma_client()
self.data = get_shared_data_client()
# ----- checkpoint helpers -----
def _save_candidates(self, candidates: Dict[str, int]) -> None:
"""Persist Phase 2 candidates so a restart can skip Phase 1+2."""
try:
self.db.set_pool_state_json(_CANDIDATES_KEY, {
"candidates": candidates,
"saved_at": time.time(),
})
except Exception as e:
logger.debug(f"[pool] save candidates checkpoint failed: {e}")
def _load_candidates(self) -> Dict[str, int]:
"""Load candidates checkpoint. Returns {} if missing/expired/invalid."""
try:
raw = self.db.get_pool_state_json(_CANDIDATES_KEY)
except Exception:
return {}
if not raw or not isinstance(raw, dict):
return {}
saved_at = raw.get("saved_at", 0)
if time.time() - saved_at > _RESUME_TTL_SECONDS:
logger.info("[pool] discarding expired candidates checkpoint")
self._clear_checkpoint()
return {}
cands = raw.get("candidates")
if not isinstance(cands, dict) or not cands:
return {}
return {str(k): int(v) for k, v in cands.items()}
def _save_progress(self, scanned: Set[str], passed: List[dict]) -> None:
"""Persist Phase 3 progress (scanned set + passed wallets)."""
try:
self.db.set_pool_state_json(_PROGRESS_KEY, {
"scanned": list(scanned),
"passed": passed,
"saved_at": time.time(),
})
except Exception as e:
logger.debug(f"[pool] save progress checkpoint failed: {e}")
def _load_progress(self) -> tuple:
"""Load Phase 3 progress. Returns (scanned_set, passed_list)."""
try:
raw = self.db.get_pool_state_json(_PROGRESS_KEY)
except Exception:
return set(), []
if not raw or not isinstance(raw, dict):
return set(), []
saved_at = raw.get("saved_at", 0)
if time.time() - saved_at > _RESUME_TTL_SECONDS:
logger.info("[pool] discarding expired progress checkpoint")
self._clear_checkpoint()
return set(), []
scanned_list = raw.get("scanned", [])
passed = raw.get("passed", [])
if not isinstance(scanned_list, list) or not isinstance(passed, list):
return set(), []
return set(scanned_list), passed
def _clear_checkpoint(self) -> None:
"""Remove both checkpoint keys after successful build."""
try:
self.db.set_pool_state_json(_CANDIDATES_KEY, {})
self.db.set_pool_state_json(_PROGRESS_KEY, {})
except Exception as e:
logger.debug(f"[pool] clear checkpoint failed: {e}")
async def build_pool_async(self, max_markets: int = 200) -> List[dict]:
"""Async build with parallelism + per-call timeout.
Supports resume: if a valid checkpoint exists from a previous
interrupted run, skips Phase 1+2 and resumes Phase 3.
Returns list of wallet dicts (already filtered by min_pnl/trades/categories).
"""
settings = self.settings
@@ -56,134 +153,245 @@ class WalletPoolBuilder:
f"(concurrency={concurrency}, request_timeout={timeout}s)"
)
# Phase 1: get top events (single call)
try:
events = await asyncio.to_thread(
self.gamma.get_active_events_by_volume, max_markets
# ----- Try resume from checkpoint -----
candidates = self._load_candidates()
scanned_before: Set[str] = set()
passed_before: List[dict] = []
if candidates:
scanned_before, passed_before = self._load_progress()
logger.info(
f"[pool] RESUME: loaded {len(candidates)} candidates from checkpoint, "
f"{len(scanned_before)} already scanned, {len(passed_before)} passed filter. "
f"Skipping Phase 1+2."
)
except Exception as e:
logger.error(f"[pool] Gamma API failed: {e}")
return []
condition_ids: Set[str] = set()
for ev in events:
for m in (ev.get("markets") or []):
cid = m.get("conditionId")
if cid and not m.get("closed"):
condition_ids.add(cid)
logger.info(f"[pool] Found {len(condition_ids)} candidate markets")
# Phase 2: parallel holder fetch
semaphore = asyncio.Semaphore(concurrency)
empty_streak = 0
async def fetch_holders(cid: str) -> List[dict]:
nonlocal empty_streak
async with semaphore:
if empty_streak >= settings.wallet_pool_backoff_emails:
logger.info(f"[pool] {empty_streak} consecutive empty responses, sleeping 15s")
await asyncio.sleep(15)
empty_streak = 0
try:
result = await asyncio.to_thread(
self.data.get_top_holders, cid, 30, timeout
)
if isinstance(result, list) and len(result) == 0:
empty_streak += 1
else:
empty_streak = 0
return result
except Exception as e:
logger.debug(f"[pool] holders failed for {cid[:10]}: {e}")
empty_streak += 1
return []
market_tasks = [fetch_holders(cid) for cid in condition_ids]
candidates: Dict[str, int] = {}
completed = 0
total = len(market_tasks)
for coro in asyncio.as_completed(market_tasks):
holders = await coro
completed += 1
# Response shape: [{"token": "...", "positions": [...]}]
for token_wrapper in holders:
for pos in (token_wrapper.get("positions") or []):
addr = pos.get("proxyWallet")
if addr:
candidates[addr] = candidates.get(addr, 0) + 1
if completed % progress_every == 0 or completed == total:
logger.info(
f"[pool] holders: scanned {completed}/{total} markets "
f"({len(candidates)} unique wallets so far)"
else:
# ----- Phase 1: get top events (single call) -----
try:
events = await asyncio.to_thread(
self.gamma.get_active_events_by_volume, max_markets
)
except Exception as e:
logger.error(f"[pool] Gamma API failed: {e}")
return []
if not candidates:
logger.warning("[pool] No candidate wallets found")
return []
condition_ids: Set[str] = set()
for ev in events:
for m in (ev.get("markets") or []):
cid = m.get("conditionId")
if cid and not m.get("closed"):
condition_ids.add(cid)
logger.info(f"[pool] Found {len(condition_ids)} candidate markets")
logger.info(f"[pool] Found {len(candidates)} candidate wallets")
# ----- Phase 2: parallel holder fetch -----
semaphore = asyncio.Semaphore(concurrency)
empty_streak = 0
async def fetch_holders(cid: str) -> List[dict]:
nonlocal empty_streak
async with semaphore:
if empty_streak >= settings.wallet_pool_backoff_emails:
logger.info(f"[pool] {empty_streak} consecutive empty responses, sleeping 15s")
await asyncio.sleep(15)
empty_streak = 0
try:
result = await asyncio.to_thread(
self.data.get_top_holders, cid, 30, timeout
)
if isinstance(result, list) and len(result) == 0:
empty_streak += 1
else:
empty_streak = 0
return result
except Exception as e:
logger.debug(f"[pool] holders failed for {cid[:10]}: {e}")
empty_streak += 1
return []
market_tasks = [fetch_holders(cid) for cid in condition_ids]
candidates = {}
completed = 0
total = len(market_tasks)
for coro in asyncio.as_completed(market_tasks):
holders = await coro
completed += 1
# Response shape: [{"token": "...", "positions": [...]}]
for token_wrapper in holders:
for pos in (token_wrapper.get("positions") or []):
addr = pos.get("proxyWallet")
if addr:
candidates[addr] = candidates.get(addr, 0) + 1
if completed % progress_every == 0 or completed == total:
logger.info(
f"[pool] holders: scanned {completed}/{total} markets "
f"({len(candidates)} unique wallets so far)"
)
if not candidates:
logger.warning("[pool] No candidate wallets found")
return []
logger.info(f"[pool] Found {len(candidates)} candidate wallets")
# Persist candidates checkpoint so a restart can skip Phase 1+2
self._save_candidates(candidates)
# ----- Phase 3: parallel position + trade + closed-position profile fetch -----
# Ensure semaphore exists (it was created in Phase 2 for the non-resume path)
if "semaphore" not in locals():
semaphore = asyncio.Semaphore(concurrency)
# Phase 3: parallel position profile fetch
async def fetch_wallet(addr: str) -> dict:
async with semaphore:
cutoff = int(time.time()) - settings.wallet_pnl_window_days * 86400
try:
positions = await asyncio.to_thread(
self.data.get_positions, addr, 500, timeout * 2
# Pool building: use half the configured max_pages for
# speed. PnL only needs to be accurate enough to pass
# the WALLET_PNL_MIN_USD filter; precise PnL is computed
# later in the bayesian updater (which uses full max_pages).
pool_max_pages = max(5, settings.closed_positions_max_pages // 2)
positions, trades, closed = await asyncio.gather(
asyncio.to_thread(self.data.get_positions, addr, 500, timeout * 2),
asyncio.to_thread(self.data.get_trades, addr, 1000, start_ts=cutoff),
asyncio.to_thread(
self.data.get_closed_positions_since, addr, cutoff,
pool_max_pages, 50, timeout, # page_size, per-request timeout
),
return_exceptions=True,
)
except Exception as e:
logger.debug(f"[pool] positions failed for {addr[:10]}: {e}")
logger.debug(f"[pool] fetch failed for {addr[:10]}: {e}")
positions = []
trades = []
closed = []
cash_pnl = sum(float(p.get("cashPnl") or 0) for p in positions)
realized_pnl = sum(
float(p.get("realizedPnl") or 0) for p in positions
)
if isinstance(positions, Exception):
positions = []
if isinstance(trades, Exception):
trades = []
if isinstance(closed, Exception):
closed = []
# P3-11: market-maker detection. A market maker holds both
# Yes AND No on the same conditionId (providing liquidity on
# both sides). Their trades are not directional and following
# them is noise. We detect by grouping positions by
# conditionId and counting markets where the wallet has 2+
# distinct outcomes. If >10% of markets are double-sided,
# exclude the wallet entirely.
if settings.exclude_market_makers and positions:
cid_outcomes: Dict[str, set] = {}
for p in positions:
pcid = p.get("conditionId")
pout = p.get("outcome")
if pcid and pout:
cid_outcomes.setdefault(pcid, set()).add(pout)
double_sided = sum(1 for outs in cid_outcomes.values() if len(outs) >= 2)
if cid_outcomes and double_sided / len(cid_outcomes) > 0.10:
logger.debug(
f"[pool] {addr[:10]} excluded as market maker "
f"({double_sided}/{len(cid_outcomes)} markets double-sided)"
)
return None
# P3-12: discount unrealized cashPnl. Polymarket mid-prices
# often overstate actual fillable value due to thin liquidity
# and bid-ask spread. We only count a fraction (default 50%)
# of unrealized PnL when scoring wallet health, while counting
# realized PnL at full value.
cash_pnl_raw = sum(float(p.get("cashPnl") or 0) for p in positions)
cash_pnl = cash_pnl_raw * settings.cash_pnl_discount
realized_pnl = sum(float(c.get("realizedPnl") or 0) for c in closed)
total_pnl = cash_pnl + realized_pnl
trades = len(positions)
categories = len({
p.get("eventSlug") or p.get("slug")
for p in positions
if p.get("eventSlug") or p.get("slug")
})
# Trade count: deduplicate by transactionHash from /trades
tx_hashes = set()
for t in trades:
txh = t.get("transactionHash")
if txh:
tx_hashes.add(txh)
trades_count = len(tx_hashes) if tx_hashes else len(trades)
# Categories: merge slugs from /trades + /positions
slugs = set()
for t in trades:
s = t.get("eventSlug") or t.get("slug")
if s:
slugs.add(s)
for p in positions:
s = p.get("eventSlug") or p.get("slug")
if s:
slugs.add(s)
categories = len(slugs)
score = compute_health_score(
pnl_30d=total_pnl,
total_pnl=total_pnl,
trades=trades,
pnl=total_pnl,
trades=trades_count,
categories=categories,
min_trades=settings.wallet_min_trades,
)
return {
"address": addr,
"source": "top_holders",
"pnl_30d_usd": total_pnl,
"pnl_total_usd": total_pnl,
"trades_count": trades,
"cash_pnl_raw": cash_pnl_raw,
"cash_pnl_discounted": cash_pnl,
"realized_pnl": realized_pnl,
"trades_count": trades_count,
"categories_count": categories,
"health_score": score,
"credibility": settings.bayesian_prior_skill,
"last_seen_at": datetime.now().isoformat(),
"last_seen_at": datetime.now(timezone.utc).isoformat(),
}
wallet_tasks = [fetch_wallet(addr) for addr in candidates]
wallets: List[dict] = []
# Resume: skip wallets already scanned in a previous run.
# Otherwise scan all candidates.
if scanned_before:
remaining = [a for a in candidates if a not in scanned_before]
logger.info(
f"[pool] RESUME: {len(remaining)} wallets remaining to scan "
f"(skipping {len(scanned_before)} already scanned)"
)
# Start from previously-passed wallets so they aren't lost
wallets = list(passed_before)
else:
remaining = list(candidates)
wallets = []
wallet_tasks = [fetch_wallet(addr) for addr in remaining]
completed = 0
total = len(wallet_tasks)
scanned_now: Set[str] = set()
for coro in asyncio.as_completed(wallet_tasks):
wallet = await coro
completed += 1
# fetch_wallet returns None when the wallet is excluded (e.g.
# market-maker detection). Skip those without crashing.
if wallet is None:
continue
scanned_now.add(wallet["address"])
if (
wallet["pnl_total_usd"] >= settings.wallet_pnl_min_usd
and wallet["trades_count"] >= settings.wallet_min_trades
and wallet["categories_count"] >= settings.wallet_min_categories
):
wallets.append(wallet)
# Periodic progress log
if completed % progress_every == 0 or completed == total:
logger.info(
f"[pool] profiles: scanned {completed}/{total} wallets "
f"({len(wallets)} passed filter so far)"
)
# Checkpoint: persist progress every _PERSIST_EVERY wallets so
# an interrupt doesn't lose all work. Also persist on the last
# wallet before clearing the checkpoint below.
if completed % _PERSIST_EVERY == 0 or completed == total:
all_scanned = scanned_before | scanned_now
self._save_progress(all_scanned, wallets)
# Build complete — clear checkpoints so next build starts fresh
self._clear_checkpoint()
wallets.sort(key=lambda w: w["health_score"], reverse=True)
result = wallets[:settings.wallet_pool_size]
@@ -210,8 +418,8 @@ class WalletPoolBuilder:
pass
pool = asyncio.run(self.build_pool_async(max_markets))
now = datetime.now().isoformat()
now = datetime.now(timezone.utc).isoformat()
for w in pool:
w.setdefault("added_at", now)
self.db.upsert_wallet_target(w)
return len(pool)
return len(pool)
src/utils/logger.py
+8 -8
View File
@@ -44,22 +44,22 @@ class BotLogger:
def startup(self, wallet_count: int, capital: float) -> None:
self.console.print(
f"\n[bold green]{'='*60}[/bold green]\n"
f"[bold green]📈 COPY TRADER STARTED[/bold green]\n"
f"[bold green]📈 跟单交易启动[/bold green]\n"
f"[bold green]{'='*60}[/bold green]\n"
f"[green]Target Wallets:[/green] {wallet_count}\n"
f"[green]Capital:[/green] ${capital:,.0f}\n"
f"[green]目标钱包数:[/green] {wallet_count}\n"
f"[green]本金:[/green] ${capital:,.0f}\n"
f"[bold green]{'='*60}[/bold green]\n"
)
def signal(self, market: str, side: str, strength: float, n_wallets: int) -> None:
self.console.print(
f"\n[bold magenta]{'='*60}[/bold magenta]\n"
f"[bold magenta]🎯 COPY SIGNAL[/bold magenta]\n"
f"[bold magenta]🎯 跟单信号[/bold magenta]\n"
f"[bold magenta]{'='*60}[/bold magenta]\n"
f"[green]Market:[/green] {market[:60]}\n"
f"[green]Direction:[/green] {side}\n"
f"[green]Strength:[/green] {strength:.3f}\n"
f"[green]Source Wallets:[/green] {n_wallets}\n"
f"[green]市场:[/green] {market[:60]}\n"
f"[green]方向:[/green] {side}\n"
f"[green]强度:[/green] {strength:.3f}\n"
f"[green]来源钱包:[/green] {n_wallets}\n"
f"[bold magenta]{'='*60}[/bold magenta]\n"
)