commit a1963e58eddeeb0d21cbd973e8f0b77747b1fd36 Author: gavindiaz Date: Tue Jul 14 00:37:22 2026 +0800 初步完成项目，可以监控给出入场信号 diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..b81b38f --- /dev/null +++ b/.env.example @@ -0,0 +1,67 @@ +# ============================================================ +# Polymarket Copy Trader - 配置文件 +# ============================================================ +# 复制: cp .env.example .env +# 必填项: TELEGRAM_BOT_TOKEN + TELEGRAM_CHAT_ID +# ============================================================ + +# --- 资金与仓位 --- +INITIAL_CAPITAL_USD=10000 +MAX_POSITION_PCT=0.05 +KELLY_FRACTION=0.5 + +# --- 价格防护 --- +MIN_PRICE=0.10 +MAX_PRICE=0.90 + +# --- 交易规则 --- +MIN_TRADE_SIZE_USD=500 +EXECUTION_DELAY_SECONDS=5 +ENABLE_EXECUTION=false + +# --- 钱包池（top N 监控）--- +WALLET_POOL_SIZE=100 +WALLET_PNL_MIN_USD=5000 +WALLET_MIN_TRADES=30 +WALLET_MIN_CATEGORIES=3 +WALLET_REFRESH_HOURS=24 + +# --- 贝叶斯先验 --- +BAYESIAN_PRIOR_SKILL=0.5 +BAYESIAN_DECAY_DAYS=14 + +# --- Telegram 通知 --- +TELEGRAM_ENABLED=false +TELEGRAM_BOT_TOKEN= +TELEGRAM_CHAT_ID= + +# --- Polymarket CLOB (仅启用实盘时填) --- +POLY_API_KEY= +POLY_API_SECRET= +POLY_API_PASSPHRASE= +POLY_WALLET_PRIVATE_KEY= + +# --- HTTP / 代理（按需） --- +HTTP_PROXY= + +# --- 数据源轮询 --- +USER_POLL_INTERVAL_SECONDS=30 +STREAM_WARMUP_SECONDS=30 +STREAM_MAX_TRADES_PER_WALLET=20 + +# --- 共识参数 --- +CONSENSUS_WINDOW_SECONDS=600 +CONSENSUS_MIN_WALLETS=2 +CONSENSUS_STRENGTH_THRESHOLD=0.4 +WALLET_DEBOUNCE_SECONDS=600 +MIN_CREDIBILITY=0.3 +CREDIBILITY_UPDATE_MINUTES=60 + +# --- 调试日志 --- +DEBUG_LOG_API_PAYLOADS=false + +# --- 数据库 --- +DB_PATH=data/copytrader.db + +# --- 日志 --- +LOG_LEVEL=INFO diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..4d9cfd4 --- /dev/null +++ b/.gitignore @@ -0,0 +1,38 @@ +# Environments +.env +!.env.example + +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +*.egg-info/ +dist/ +build/ +.eggs/ +*.egg +.venv/ +venv/ +env/ + +# IDE +.idea/ +.vscode/ +*.swp +*.swo +*~ + +# OS +.DS_Store +Thumbs.db + +# Docker +.docker/ + +# Reports/data +reports/ +data/ + +# Logs +*.log \ No newline at end of file diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..359a100 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,16 @@ +FROM python:3.12-slim + +WORKDIR /app + +RUN apt-get update && apt-get install -y --no-install-recommends \ + gcc \ + && rm -rf /var/lib/apt/lists/* + +COPY pyproject.toml . +RUN pip install --no-cache-dir . + +COPY src/ src/ + +RUN mkdir -p data reports + +CMD ["python", "-m", "src.main", "run"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..91ec8a9 --- /dev/null +++ b/README.md @@ -0,0 +1,71 @@ +# Polymarket Copy Trader (Project B) + +**Quantitative copy-trading bot for Polymarket** — follows the top 1% of profitable wallets using Bayesian credibility updates + Kelly position sizing + Favorite-Longshot bias correction. + +## What's Different from Project A (Whale Watcher) + +| Aspect | Project A (whale-watcher) | Project B (copy-trader) | +|---|---|---| +| Goal | Detect "insider" trades via LLM | Mirror statistically profitable wallets | +| LLM cost | $0.5-2 per signal | $0 (all math, no LLM) | +| Win rate evidence | 63.5% on 250 signals | Expected ≥55% from top-1% PnL data | +| Latency | Asynchronous | Real-time WebSocket | + +## Phase Plan + +- ✅ **Phase 0** (current): foundation, pool builder, Telegram notifier +- ⏳ **Phase 1**: User WebSocket subscriber (one per wallet, get every trade in real-time) +- ⏳ **Phase 2**: Signal aggregator (Softmax + Bayesian update + Kelly sizer) +- ⏳ **Phase 3**: CLOB trader (optional, off by default — manual first) +- ⏳ **Phase 4**: Dashboard (FastAPI for P&L tracking) + +## Architecture (target) + +``` +Data API ──┐ + ├─→ Wallet Pool Builder ──→ SQLite (top 100 wallets) + │ + └─→ Health score = f(PnL, trades, category diversity) + ↓ + User WebSocket (User channel) per wallet + ↓ + On each trade → check eligibility (size, price) + ↓ + Bayesian credibility update + ↓ + Softmax aggregation across wallets + ↓ + Kelly sizing + Favorite-Longshot correction + ↓ + Telegram notification + optional CLOB execution +``` + +## Running + +```bash +# Local +python -m src.main run + +# Docker (cloud) +docker compose build --no-cache && docker compose up -d +``` + +## Configuration + +Copy `.env.example` to `.env` and fill in: +- Polymarket Gamma + Data API (no auth needed for reads) +- Telegram bot token + chat ID for notifications +- CLOB credentials (only if enabling live execution) +- Wallet pool criteria (PnL minimum, trade count, etc.) + +## Why this will (probably) make money + +- Polymarket data: **top 1% users capture 84% of all profits** +- Their edge = identifying mispriced contracts, NOT insider info +- Following their trades = systematic exposure to that edge +- Bayesian updating = self-correcting when a wallet's skill degrades +- Favorite-Longshot correction = avoid 70.8% user loss pattern + +## Disclaimer + +Backtesting not yet performed. Start with `ENABLE_EXECUTION=false` and observe signals via Telegram for 2-4 weeks before any live trading. diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..a784e0c --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,30 @@ +services: + trader: + build: . + restart: unless-stopped + env_file: + - .env + volumes: + - ./data:/app/data + - ./reports:/app/reports + logging: + driver: "json-file" + options: + max-size: "10m" + max-file: "3" + + dashboard: + build: . + restart: unless-stopped + command: ["python", "-m", "src.main", "dashboard", "--host", "0.0.0.0", "--port", "8518"] + env_file: + - .env + ports: + - "8518:8518" + volumes: + - ./data:/app/data + logging: + driver: "json-file" + options: + max-size: "10m" + max-file: "3" diff --git a/polymarket_api_research.md b/polymarket_api_research.md new file mode 100644 index 0000000..ef73168 --- /dev/null +++ b/polymarket_api_research.md @@ -0,0 +1,371 @@ +# Polymarket API 完整调研报告 + +> 数据来源：官方文档 https://docs.polymarket.com/、Polymarket/agent-skills GitHub、polymarkets.co.il 教程 (2026-07) +> 调研日期：2026-07-13 + +--- + +## 0. 总体架构 + +Polymarket 将数据拆分为 **4 个 REST API + 3 个 WebSocket 频道**，前 3 个完全公开（无需 API Key）： + +| API | Base URL | 鉴权 | 用途 | +|---|---|---|---| +| **Gamma API** | `https://gamma-api.polymarket.com` | 公开 | 市场元数据、发现、搜索、标签 | +| **CLOB API** | `https://clob.polymarket.com` | 读公开 / 写需 L2 | 订单簿、价格、深度、历史价 | +| **Data API** | `https://data-api.polymarket.com` | 公开 | 用户持仓、交易、活跃度、Holder、OI | +| **Bridge API** | `https://bridge.polymarket.com` | 钱包 | 充提（fun.xyz 代理） | +| **RTDS WebSocket** | `wss://ws-subscriptions-clob.polymarket.com/ws/{market,user}` + `wss://sports-api.polymarket.com/ws` | 公开 / 鉴权 | 实时订单簿、成交、用户事件 | + +官方 SDK：**Python (`py-clob-client`)**、**TypeScript (`@polymarket/clob-client`)**、**Rust (`rs-clob-client`)**。第三方封装很多（`polymarket-pandas`、`polybricks-sdk`、`almanak-sdk` 等）。 + +--- + +## 1. 数据模型（核心概念） + +``` +Event（事件） ──┬── Market A ── Yes token / No token (条件 ID = conditionId) + ├── Market B ── Yes token / No token + └── Market N +``` + +- **Event**：顶级问题（如 "2024 大选谁赢？"）。每个 event 含 ≥1 个 market。 +- **Market**：实际可交易的二元/多元合约，对应一对 ERC-1155 `token_id`。 +- **核心 ID**： + - `conditionId` — 市场唯一条件 ID（0x + 64 hex） + - `questionID` — UMA 解析用的哈希 + - `tokenID / asset_id` — Yes/No 各自的 ERC-1155 token ID（**下单/查订单簿都用它**） + - `slug` — URL 友好标识 +- **价格含义**：`outcomePrices` 是隐含概率，二元市场 `[Yes, No]` 数值相加 ≈ 1。 +- **市场类型**：`negRisk=true` → 多结果互斥事件；`enableOrderBook=true` → 可在 CLOB 交易。 + +--- + +## 2. Gamma API — 市场发现与元数据 + +> 无需鉴权，REST。 + +### 2.1 端点列表 + +| 端点 | 用途 | +|---|---| +| `GET /events` | 列出 events（带过滤、分页） | +| `GET /events/{id}` | 单个 event | +| `GET /events/keyset` | keyset 分页版本（用于游标分页） | +| `GET /markets` | 列出 markets | +| `GET /markets/{id}` | 单个 market | +| `GET /public-search` | 全文搜索 events/markets/profiles | +| `GET /tags` | 标签（分类）列表 | +| `GET /series` | 系列（分组 event） | +| `GET /sports` | 体育元数据 | +| `GET /teams` | 球队 | +| `GET /comments` | 评论（部分端点） | + +### 2.2 `/events` 关键过滤参数 + +| 参数 | 说明 | +|---|---| +| `slug` | URL slug 精确匹配 | +| `id` | event id | +| `tag_id` | 标签 ID（可多次传入，支持 `related_tags=true`） | +| `tag_slug` | slug 形式过滤 | +| `exclude_tag_id` | 排除的标签 | +| `series_id` | 系列 ID（体育赛事分组） | +| `active` | 是否上架 | +| `closed` | 是否已结算（默认 `false`） | +| `archived` | 是否归档 | +| `new`, `featured`, `restricted` | 状态布尔 | +| `live` | 是否 live | +| `order` | `volume_24hr` / `volume` / `liquidity` / `start_date` / `end_date` / `competitive` / `closed_time` | +| `ascending` | `true` / `false` | +| `limit` | 1–500，默认 20 | +| `offset` | 分页偏移 | +| `include_chat`, `include_template` | 是否包含聊天/模板字段 | + +示例： +```bash +# 24h 成交量最高的活跃事件 +curl "https://gamma-api.polymarket.com/events?active=true&closed=false&order=volume24hr&ascending=false&limit=100" + +# NBA 标签下的活动 +curl "https://gamma-api.polymarket.com/events?tag_id=745&active=true&closed=false" +``` + +### 2.3 `/markets` 关键过滤参数 + +`/markets` 支持几乎所有 `/events` 的参数，外加： + +| 参数 | 说明 | +|---|---| +| `condition_ids` / `clob_token_ids` / `question_ids` | 用 ID 精确匹配 | +| `slug` | URL slug | +| `archived`, `featured`, `new`, `restricted`, `closed`, `active` | 状态 | +| `order` / `ascending` | 排序 | +| `liquidity_num_min`, `volume_num_min` | 数值范围过滤 | +| `end_date_min`, `end_date_max` | 截止日期范围 | +| `limit` (≤500) / `offset` | 分页 | + +### 2.4 `/tags` 常用分类与 ID + +| Tag | ID | +|---|---| +| Sports（运动） | 1 | +| Politics（政治） | 2 | +| Election（选举） | 3 | +| Crypto（加密货币） | 21 | +| Bitcoin（比特币） | 100196 | +| Ethereum（以太坊） | 100383 | +| NFL | 450 | +| NBA | 745 | +| Tennis | 864 | +| Esports | 702 | +| Soccer（足球） | 1059 | +| EPL | 739 | +| UCL | 2186 | +| Middle East | 1432 | + +⚠️ 教程建议生产环境只用数值 `tag_id`，`tag_slug` 和 `?q=...` 不稳定。 + +--- + +## 3. CLOB API — 价格与订单簿 + +> 读端点公开，写端点（挂单/撤单）需 L2 钱包签名鉴权。 + +### 3.1 公共只读端点 + +| 端点 | 说明 | +|---|---| +| `GET /markets` | 列出 CLOB markets（**注意与 Gamma 的 `/markets` 不同**，是简化版） | +| `GET /markets/{conditionId}` | 单个 market 详情 | +| `GET /book?token_id=...` | 某 token 的订单簿快照 | +| `POST /books` | 批量订单簿（≤500 token） | +| `GET /price?token_id=...&side=BUY\|SELL` | 买/卖最优价 | +| `POST /prices` | 批量最优价 | +| `GET /midpoint?token_id=...` | 中间价 | +| `POST /midpoints` | 批量中间价 | +| `GET /spread?token_id=...` | 价差 | +| `POST /spreads` | 批量价差 | +| `GET /last-trade-price?token_id=...` | 最近成交价 | +| `POST /last-trades-prices` | 批量最近成交 | +| `GET /prices-history?market=...` | **历史价格** | +| `POST /prices-history` | 批量历史价格 | + +### 3.2 历史价格 `/prices-history` 参数 + +| 参数 | 必填 | 说明 | +|---|---|---| +| `market` | ✅ | token_id（asset_id） | +| `startTs` | ❌ | Unix 秒（与 `interval` 互斥） | +| `endTs` | ❌ | Unix 秒 | +| `interval` | ❌ | `max` / `all` / `1m` / `1w` / `1d` / `6h` / `1h` | +| `fidelity` | ❌ | 精度（分钟），默认 1 | + +返回 `[{ "t": timestamp, "p": price }, ...]`。 + +### 3.3 `/markets` (CLOB) 参数 + +- 支持 `next_cursor` 游标分页（新版上限 limit=100，需用 `after_cursor`） +- 过滤字段：`active`, `closed`, `archived`, `condition_ids`, `clob_token_ids`, `question_ids`, `slug`, `tags`, `order`, `ascending` + +### 3.4 鉴权写入端点（L2） + +挂单、撤单、查用户订单、心跳、API key 创建等。需用 `POLY_API_KEY / POLY_ADDRESS / POLY_SIGNATURE / POLY_PASSPHRASE / POLY_TIMESTAMP` 头部 HMAC 签名。 + +--- + +## 4. Data API — 持仓、交易、Holder、OI + +> 全部公开。 + +### 4.1 端点列表 + +| 端点 | 关键参数 | +|---|---| +| `GET /positions` | `user` (0x 地址, 必填), `market` (conditionId 数组), `eventId`, `sizeThreshold`, `redeemable`, `mergeable`, `limit` (≤500), `offset` (≤10000), `sortBy` (`CURRENT`/`INITIAL`/`TOKENS`/`CASHPNL`/`PERCENTPNL`/`TITLE`/`RESOLVING`/`PRICE`/`AVGPRICE`), `sortDirection` (`ASC`/`DESC`), `title` | +| `GET /closed-positions` | `user` (必填), `market[]`, `eventId[]`, `title`, `limit` (≤50), `offset` (≤100000), `sortBy` (`REALIZEDPNL`/`TITLE`/`PRICE`/`AVGPRICE`/`TIMESTAMP`), `sortDirection` | +| `GET /activity` | `user` (必填), `market[]`, `eventId[]`, `type` (`TRADE`/`SPLIT`/`MERGE`/`REDEEM`/`REWARD`/`CONVERSION`/`DEPOSIT`/`WITHDRAWAL`/`YIELD`/`MAKER_REBATE`/`TAKER_REBATE`/`REFERRAL_REWARD`), `side` (`BUY`/`SELL`), `start`/`end` Unix 秒, `sortBy` (`TIMESTAMP`/`TOKENS`/`CASH`), `sortDirection`, `limit` (≤500), `offset` | +| `GET /v1/market-positions` | **按市场查 top 持仓人**：`market` (conditionId, 必填), `user`, `status` (`OPEN`/`CLOSED`/`ALL`), `sortBy` (`TOKENS`/`CASH_PNL`/`REALIZED_PNL`/`TOTAL_PNL`), `sortDirection`, `limit` (≤500), `offset` | +| `GET /value?user=...` | 用户所有持仓的总价值（美元） | +| `GET /oi` | 全市场未平仓量（按市场聚合） | +| `GET /holders?market={conditionId}` | 某市场的 top 持仓人 | +| `GET /trades` | 全市场成交流（见下） | + +### 4.2 `/trades`（成交历史）参数 + +| 参数 | 说明 | +|---|---| +| `market` | conditionId | +| `asset_id` | token ID | +| `maker_address` | 必填，按 maker 过滤 | +| `before`, `after` | Unix 秒时间窗 | +| `next_cursor` | base64 游标分页 | +| `id` | 单笔 trade ID | + +鉴权：需 L2 API key 头。返回字段包括 `taker_order_id`、`asset_id`、`side`、`size`、`fee_rate_bps`、`price`、`status`、`match_time`、`outcome`、`owner`、`maker_address`、`transaction_hash`、`trader_side`、`maker_orders[]`。 + +### 4.3 返回字段（持仓示例） + +``` +proxyWallet, asset, conditionId, size, avgPrice, initialValue, currentValue, +cashPnl, percentPnl, totalBought, realizedPnl, percentRealizedPnl, curPrice, +redeemable, mergeable, title, slug, icon, eventSlug, outcome, outcomeIndex, +oppositeOutcome, oppositeAsset, endDate, negativeRisk +``` + +### 4.4 其他 Data 端点 + +- `GET /leaderboard` — 排行榜 +- `GET /profiles/{address}` — 公开档案 +- `GET /comments/by-user/{address}` — 用户评论 +- `GET /builder-leaderboard` — builder 排行榜 +- `GET /v1/accounting/snapshot` — 会计快照下载（ZIP/CSV） +- `GET /v1/positions/combos`, `/v1/activity/combos` — 组合仓 + +--- + +## 5. WebSocket — 实时数据 + +| 频道 | Endpoint | 鉴权 | 订阅方式 | +|---|---|---|---| +| Market | `wss://ws-subscriptions-clob.polymarket.com/ws/market` | 否 | 按 `assets_ids[]`（token_id） | +| User | `wss://ws-subscriptions-clob.polymarket.com/ws/user` | **是** | 按 `markets[]`（conditionId），省略=全部 | +| Sports | `wss://sports-api.polymarket.com/ws` | 否 | 无需订阅 | + +### 5.1 Market 频道事件类型 + +| 事件 | 触发 | +|---|---| +| `book` | 订阅时 + 影响订单簿的成交 | +| `price_change` | 挂单/撤单 | +| `last_trade_price` | 成交 | +| `tick_size_change` | 价格越过 0.96 / 0.04（**做市必听**） | +| `best_bid_ask` | 需 `custom_feature_enabled=true` | +| `new_market` | 新市场创建 | +| `market_resolved` | 结算 | + +订阅消息： +```json +{"assets_ids":["TOKEN_ID"], "type":"market", "custom_feature_enabled":true} +``` + +### 5.2 User 频道事件 + +- `trade` — 成交生命周期（MATCHED / MINED / CONFIRMED / RETRYING / FAILED） +- `order` — 订单生命周期（PLACEMENT / UPDATE / CANCELLATION） + +### 5.3 心跳 + +- Market/User：**客户端每 10 秒发 `PING`** +- Sports：**服务端每 5 秒发 `ping`，客户端必须在 10 秒内回 `pong`**，否则断连 + +--- + +## 6. 子图 / 链上数据（Goldsky Subgraph） + +通过 GraphQL 查询 Polymarket on-chain 数据： + +| 子图 | 内容 | +|---|---| +| Positions | 用户 token 余额 | +| Orders | 订单簿和成交事件 | +| Activity | Split / Merge / Redeem | +| OpenInterest | 市场 OI | +| PNL | 用户盈亏 | + +端点（Goldsky 托管）： +``` +https://api.goldsky.com/api/public/project_cl6mb8i9h0003e201j6li0diw/subgraphs/orderbook-subgraph/0.0.1/gn +``` + +--- + +## 7. 速率限制 & 缓存建议 + +- Gamma / Data API（Cloudflare 前置）： + - 同一 IP **≤30 req/s** 持续 → 安全 + - **100+ req/s** 突发 → 偶发 429（重试可恢复） + - **500 req/s** 持续 → 限流 / 封禁 10–60s + - 响应头 `Cache-Control` 通常 30–60s +- 推荐：单 fetcher 进程 + 本地 LRU 缓存（30s TTL），多个消费者读取本地副本 +- 大量拉取时建议加 1 fetcher × N consumers，避免多 fetcher 抢配额 + +--- + +## 8. 常用查询模式速查 + +### 8.1 找"现在最热"的市场 +```bash +curl "https://gamma-api.polymarket.com/events?active=true&closed=false&order=volume24hr&ascending=false&limit=100" +``` + +### 8.2 查某个分类所有活跃市场 +```bash +# 政治类（Politics tag = 2） +curl "https://gamma-api.polymarket.com/events?tag_id=2&active=true&closed=false&order=volume&ascending=false" +``` + +### 8.3 拿单个市场的实时报价 +```bash +# 1. 用 slug 查 conditionId / clobTokenIds +curl "https://gamma-api.polymarket.com/events?slug=fed-decision-in-october" +# 2. 用 token_id 查订单簿 +curl "https://clob.polymarket.com/book?token_id=" +# 3. 查历史价 +curl "https://clob.polymarket.com/prices-history?market=&interval=1d&fidelity=60" +``` + +### 8.4 监控一个鲸鱼地址 +```bash +# 当前持仓 +curl "https://data-api.polymarket.com/positions?user=0x...&sizeThreshold=1&limit=500&sortBy=CURRENT" +# 历史成交 +curl "https://data-api.polymarket.com/activity?user=0x...&type=TRADE&limit=500" +# 订阅 User WebSocket 实时 trade/order +``` + +### 8.5 找热门市场的 top 持有人 +```bash +curl "https://data-api.polymarket.com/v1/market-positions?market=&status=OPEN&sortBy=TOTAL_PNL&sortDirection=DESC&limit=50" +``` + +--- + +## 9. 注意事项 / 坑 + +1. **Gamma 的 `outcomes` / `outcomePrices` / `clobTokenIds` 字段是 JSON 字符串而非数组**，需 `json.loads()` 再用。 +2. **不要用 `?q=...` 搜索**：`/events` 不支持自由文本，文档显式说"silently returns default ordering"，请用 `/public-search` 或 slug。 +3. **`tag_slug` 不稳定**：教程建议生产环境只用数值 `tag_id`。 +4. **排序与标签同时过滤可能返回空**：tag 下市场太少 + `volume24hr desc` 时会空，需要 over-fetch 客户端再过滤。 +5. **WebSocket 必须立刻发订阅**：连接后未立刻发订阅消息会被立刻断开。 +6. **CLOB 价格用 token_id 不是 slug**：先 Gamma 拿 `clobTokenIds` 再查 CLOB。 +7. **WSS 心跳**：Market/User 是**客户端发 PING**；Sports 是**服务端发 ping**。两者别搞反。 +8. **`tick_size_change` 是做市关键事件**：tick 改了还用旧值会被拒单。 +9. **`/markets` 在 CLOB 与 Gamma 两个端点都有**，含义不同： + - `clob.polymarket.com/markets`：交易用的简化市场（limit 上限 100，需游标） + - `gamma-api.polymarket.com/markets`：元数据 + 内嵌 markets（limit 上限 500，offset） +10. **地理限制**：某些地区会被 Cloudflare 屏蔽，必要时用代理。 + +--- + +## 10. 常用分类/标签 ID 速查 + +| 分类 | Tag ID | +|---|---| +| Sports | 1 | +| Politics | 2 | +| Election | 3 | +| Crypto | 21 | +| Bitcoin | 100196 | +| Ethereum | 100383 | +| Pop Culture / Entertainment | （通过 `/tags` 动态查询） | +| Middle East | 1432 | +| NBA | 745 | +| NFL | 450 | +| Tennis | 864 | +| Esports | 702 | +| Soccer | 1059 | +| EPL | 739 | +| UCL | 2186 | + +完整列表通过 `GET https://gamma-api.polymarket.com/tags?limit=500` 获取。 \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..b3d3e65 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,39 @@ +[build-system] +requires = ["setuptools>=61.0", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "polymarket-copy-trader" +version = "0.1.0" +description = "Quantitative copy-trading bot for Polymarket — follows top profitable wallets using Bayesian + Kelly sizing" +readme = "README.md" +requires-python = ">=3.10" +license = {text = "MIT"} +dependencies = [ + "httpx>=0.27.0", + "websockets>=13.0", + "requests>=2.28.0", + "pydantic>=2.0.0", + "pydantic-settings>=2.0.0", + "python-dotenv>=1.0.0", + "rich>=13.0.0", + "typer>=0.9.0", + "fastapi>=0.100.0", + "uvicorn>=0.20.0", + "numpy>=1.24.0", + "python-telegram-bot>=21.0", +] + +[project.optional-dependencies] +dev = [ + "pytest>=7.0.0", + "pytest-asyncio>=0.23.0", + "ruff>=0.1.0", +] + +[project.scripts] +copy-trader = "src.main:main" + +[tool.ruff] +line-length = 100 +target-version = "py310" diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/config/__init__.py b/src/config/__init__.py new file mode 100644 index 0000000..2c2189e --- /dev/null +++ b/src/config/__init__.py @@ -0,0 +1,3 @@ +from src.config.settings import Settings, get_settings + +__all__ = ["Settings", "get_settings"] diff --git a/src/config/settings.py b/src/config/settings.py new file mode 100644 index 0000000..6dbadaf --- /dev/null +++ b/src/config/settings.py @@ -0,0 +1,89 @@ +"""Application settings for project B (copy-trader).""" +import os +from functools import lru_cache + +from dotenv import load_dotenv +from pydantic import Field +from pydantic_settings import BaseSettings + + +class Settings(BaseSettings): + """Settings loaded from .env.""" + + # Capital & execution + initial_capital_usd: float = Field(default=10000.0, alias="INITIAL_CAPITAL_USD") + max_position_pct: float = Field(default=0.05, alias="MAX_POSITION_PCT") + kelly_fraction: float = Field(default=0.5, alias="KELLY_FRACTION") + + # Price guardrails + min_price: float = Field(default=0.10, alias="MIN_PRICE") + max_price: float = Field(default=0.90, alias="MAX_PRICE") + + # Trading rules + min_trade_size_usd: float = Field(default=500.0, alias="MIN_TRADE_SIZE_USD") + execution_delay_seconds: float = Field(default=5.0, alias="EXECUTION_DELAY_SECONDS") + enable_execution: bool = Field(default=False, alias="ENABLE_EXECUTION") + + # 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_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") + + # 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") + + # 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") + + # Aggregator (Phase 2) + consensus_window_seconds: int = Field(default=600, alias="CONSENSUS_WINDOW_SECONDS") + consensus_min_wallets: int = Field(default=2, alias="CONSENSUS_MIN_WALLETS") + 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") + + # Credibility update loop + credibility_update_minutes: int = Field(default=60, alias="CREDIBILITY_UPDATE_MINUTES") + + # 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_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_backoff_emails: int = Field(default=30, alias="WALLET_POOL_BACKOFF_EMA") # consecutive empty responses → sleep + + # Telegram notifications + telegram_enabled: bool = Field(default=False, alias="TELEGRAM_ENABLED") + telegram_bot_token: str = Field(default="", alias="TELEGRAM_BOT_TOKEN") + telegram_chat_id: str = Field(default="", alias="TELEGRAM_CHAT_ID") + + # Polymarket CLOB credentials (for live execution) + poly_api_key: str = Field(default="", alias="POLY_API_KEY") + poly_api_secret: str = Field(default="", alias="POLY_API_SECRET") + poly_api_passphrase: str = Field(default="", alias="POLY_API_PASSPHRASE") + poly_wallet_private_key: str = Field(default="", alias="POLY_WALLET_PRIVATE_KEY") + + # HTTP + http_proxy: str = Field(default="", alias="HTTP_PROXY") + db_path: str = Field(default="data/copytrader.db", alias="DB_PATH") + + # Logging + log_level: str = Field(default="INFO", alias="LOG_LEVEL") + + class Config: + env_file = ".env" + env_file_encoding = "utf-8" + extra = "ignore" + + +@lru_cache +def get_settings() -> Settings: + load_dotenv() + return Settings() diff --git a/src/dashboard.py b/src/dashboard.py new file mode 100644 index 0000000..ef8affe --- /dev/null +++ b/src/dashboard.py @@ -0,0 +1,161 @@ +"""FastAPI dashboard for the copy-trader. + +Endpoints: + /api/stats — overall counters + /api/wallets — top-priority wallet pool + /api/signals/recent — recent consensus signals + /api/health — liveness check +""" +import logging +from typing import Optional + +from fastapi import FastAPI, Query +from fastapi.responses import HTMLResponse + +from src.config import get_settings +from src.db.database import CopyTraderDatabase + +logger = logging.getLogger(__name__) +app = FastAPI(title="Polymarket Copy Trader Dashboard") + + +def _get_db() -> CopyTraderDatabase: + settings = get_settings() + return CopyTraderDatabase(settings.db_path) + + +@app.get("/api/stats") +def api_stats(): + db = _get_db() + return 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() + 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) + + +@app.get("/api/health") +def api_health(): + return {"status": "ok"} + + +HTML_TEMPLATE = """ + + + + +Polymarket Copy Trader Dashboard + + + + +

Polymarket Copy Trader

+

Live signal & wallet dashboard · auto-refresh 30s

+ +
Loading...
+ + + + + +""" + + +@app.get("/", response_class=HTMLResponse) +def dashboard_page(): + return HTML_TEMPLATE diff --git a/src/db/__init__.py b/src/db/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/db/database.py b/src/db/database.py new file mode 100644 index 0000000..b3029c0 --- /dev/null +++ b/src/db/database.py @@ -0,0 +1,229 @@ +"""SQLite storage for wallet pool, signals, and trade executions.""" +import json +import logging +import sqlite3 +from datetime import datetime +from pathlib import Path +from typing import List, Optional + +logger = logging.getLogger(__name__) + + +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) + 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 _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) + ) + """) + + # ----- 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, + )) + + def get_all_wallet_targets(self) -> List[dict]: + with self._get_conn() as conn: + rows = 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() + return [r["address"] for r in rows] + + def get_wallet_target(self, address: str) -> Optional[dict]: + with self._get_conn() as conn: + row = 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), + ) + + def get_signals_count(self) -> int: + with self._get_conn() as conn: + return 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] + + def get_executed_count(self) -> int: + with self._get_conn() as conn: + return 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 + + 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,) + ) + + def get_recent_signals(self, limit: int = 50) -> List[dict]: + with self._get_conn() as conn: + rows = 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 + + # ----- 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( + "SELECT COUNT(*) FROM copy_signals WHERE executed=1" + ).fetchone()[0] + return { + "wallet_count": total, + "total_signals": sigs, + "executed_signals": executed, + } diff --git a/src/main.py b/src/main.py new file mode 100644 index 0000000..4a7eef4 --- /dev/null +++ b/src/main.py @@ -0,0 +1,284 @@ +"""Copy trader main entry point (Typer CLI). + +Commands: + - run : orchestrator (pool refresh + trade stream + bayesian + telegram) + - pool : one-shot rebuild of wallet pool + - stats : DB stats + - dashboard : start FastAPI dashboard (port 8518) + - test-stream : 1-minute test that polls top 3 wallets and prints new trades +""" +import asyncio +import signal as sys_signal +import sys +from datetime import datetime +from pathlib import Path +from typing import Optional + +import typer + +from src.config import get_settings +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.telegram import TelegramNotifier +from src.services.user_stream import UserTradeStream +from src.services.wallet_pool import WalletPoolBuilder +from src.utils.logger import BotLogger, setup_logging + +app = typer.Typer(help="Polymarket Copy Trader — follow top wallets via consensus") +logger = BotLogger() + + +class CopyTrader: + """Main orchestrator.""" + + def __init__(self): + self.settings = get_settings() + self.db = CopyTraderDatabase(self.settings.db_path) + self.sizer = KellySizer() + self.pool_builder = WalletPoolBuilder(self.db) + self.aggregator = SignalAggregator(self.db, sizer=self.sizer) + self.bayes = BayesianUpdater(self.db, self.aggregator) + self.stream = UserTradeStream(self.db) + self.notifier = TelegramNotifier() + self._running = False + self._tasks = [] + + async def run(self) -> None: + self._running = True + await self._bootstrap() + + stats = self.db.get_stats() + logger.startup( + wallet_count=stats["wallet_count"], + 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}" + ) + + 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"), + ] + try: + await asyncio.gather(*self._tasks) + except asyncio.CancelledError: + logger.info("Cancelled") + + async def _bootstrap(self) -> None: + logger.info("Bootstrapping...") + await self.notifier.start() + self.aggregator.load_credibilities() + if self.aggregator.wallet_credibility == {}: + logger.info("No wallets in pool — running initial pool build") + 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"Initial pool built: {len(pool)} wallets") + except Exception as e: + logger.error(f"Initial pool build failed: {e}") + await self.notifier.send_error(f"Initial pool build failed: {e}") + self.aggregator.load_credibilities() + logger.info(f"Pool loaded: {len(self.aggregator.wallet_credibility)} wallets") + + 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), + ) + 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) + + async def _trade_stream_loop(self) -> None: + logger.info("[loop] trade stream starting") + + 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]}" + ) + 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"], + ) + + 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") + while self._running: + await asyncio.sleep(seconds) + try: + await self.bayes._update_once() + except Exception as e: + logger.error(f"[loop] credibility update failed: {e}") + + async def _health_beat(self) -> None: + """Emit periodic stats for log visibility.""" + while self._running: + await asyncio.sleep(600) + 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}" + ) + + def stop(self) -> None: + self._running = False + self.stream.stop() + self.bayes.stop() + for t in self._tasks: + t.cancel() + + +_watcher: Optional[CopyTrader] = None + + +def signal_handler(signum, frame): + if _watcher: + _watcher.stop() + sys.exit(0) + + +@app.command() +def run( + debug: bool = typer.Option(False, "--debug", "-d", help="Verbose DEBUG logs"), +): + """Start the copy-trader bot.""" + global _watcher + setup_logging("DEBUG" if debug else "INFO") + + sys_signal.signal(sys_signal.SIGINT, signal_handler) + sys_signal.signal(sys_signal.SIGTERM, signal_handler) + + _watcher = CopyTrader() + try: + asyncio.run(_watcher.run()) + except KeyboardInterrupt: + pass + finally: + logger.info("Copy trader stopped") + + +@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}") + + +@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']}") + + +@app.command() +def test_stream(minutes: int = 2): + """One-shot test: poll top 3 wallets for N minutes, print new trades.""" + setup_logging("DEBUG") + async def _run(): + settings = get_settings() + db = CopyTraderDatabase(settings.db_path) + sizer = KellySizer() + agg = SignalAggregator(db, sizer=sizer) + agg.load_credibilities() + stream = UserTradeStream(db) + addrs = db.get_wallet_addresses()[:3] + print(f"Top 3 wallets: {addrs}") + + async def cb(addr, t): + print( + f" [{datetime.now().strftime('%H:%M:%S')}] " + f"{addr[:8]}... {t.get('side')} {t.get('outcome')} " + f"${float(t.get('size', 0)) * float(t.get('price', 0)):.0f} " + f"@ {float(t.get('price', 0)):.4f} " + f"[cid={t.get('conditionId', '?')[:12]}]" + ) + + original_run = stream.run + async def patched_run(on_trade): + self_ref = stream + self_ref._running = True + warmup_until = __import__('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) + except Exception as e: + print(f"poll err: {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") + + asyncio.run(_run()) + + +@app.command() +def dashboard(port: int = 8518, host: str = "0.0.0.0"): + """Start FastAPI dashboard.""" + import uvicorn + from src.dashboard import app as dashboard_app + + print(f"Dashboard starting at http://{host}:{port}") + uvicorn.run(dashboard_app, host=host, port=port, log_level="info") + + +def main(): + app() + + +if __name__ == "__main__": + main() diff --git a/src/models/__init__.py b/src/models/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/models/wallet.py b/src/models/wallet.py new file mode 100644 index 0000000..0ee029d --- /dev/null +++ b/src/models/wallet.py @@ -0,0 +1,34 @@ +"""Data models for the copy-trader.""" +from dataclasses import dataclass, field +from datetime import datetime +from typing import Optional + + +@dataclass +class WalletTarget: + """A top-wallet we are watching and may follow.""" + address: str + source: str # "top_holders", "trade_history", "leaderboard" + pnl_30d_usd: float + pnl_total_usd: float + trades_count: int + categories_count: int + health_score: float # 0-100 + credibility: float # Bayesian posterior (0-1) + last_seen_at: Optional[str] = None + added_at: Optional[str] = None + + +@dataclass +class CopySignal: + """Aggregated signal to follow top wallets on a market.""" + condition_id: str + market_question: str + side: str # "BUY" or "SELL" + outcome: str # "Yes" or "No" + entry_price: float + aggregated_strength: float # 0-1 weighted consensus + 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()) diff --git a/src/services/__init__.py b/src/services/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/services/aggregator.py b/src/services/aggregator.py new file mode 100644 index 0000000..cb42c19 --- /dev/null +++ b/src/services/aggregator.py @@ -0,0 +1,255 @@ +"""Signal aggregator: combines trades from multiple wallets into consensus signals. + +Flow: + on_trade(wallet_addr, trade) + │ + ├─ filter: price guard, size guard, credibility guard, debounce + │ + ├─ update per-market rolling accumulator + │ + ├─ if consensus reached: emit CopySignal + │ + └─ cleanup expired accumulators + +Consensus rule: + - N wallets (≥ CONSENSUS_MIN_WALLETS) agree on same direction within window + - aggregated strength = mean credibility of agreeing wallets + - emit only when (agreed_strength - opposed_strength) >= CONSENSUS_STRENGTH_THRESHOLD +""" +import json +import logging +import time +from dataclasses import dataclass, field +from typing import 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.kelly import KellySizer + +logger = logging.getLogger(__name__) + + +@dataclass +class MarketAccumulator: + condition_id: str + market_question: str = "" + outcome: str = "Yes" + buy_strength: float = 0.0 + sell_strength: float = 0.0 + contributors_buy: List[dict] = field(default_factory=list) + contributors_sell: List[dict] = field(default_factory=list) + last_price: float = 0.0 + last_trade_at: float = 0.0 + window_started_at: float = 0.0 + + +class SignalAggregator: + """Detects consensus signals from a stream of wallet trades.""" + + def __init__( + self, + db: CopyTraderDatabase, + sizer: Optional[KellySizer] = None, + data: Optional[DataAPIClient] = None, + ): + self.db = db + self.settings = get_settings() + self.sizer = sizer or KellySizer() + self.data = data or DataAPIClient() + self.accumulators: Dict[str, MarketAccumulator] = {} + self.wallet_credibility: Dict[str, float] = {} + self.wallet_debounce: Dict[str, float] = {} + self._signals_emitted = 0 + + def load_credibilities(self) -> None: + """Bulk-load credibility from DB into memory.""" + for w in self.db.get_all_wallet_targets(): + self.wallet_credibility[w["address"]] = w.get("credibility", 0.5) + logger.info( + 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 _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 + + async def on_trade(self, wallet_addr: str, trade: dict) -> Optional[dict]: + """Process a new trade. Returns a CopySignal dict if consensus reached.""" + cid = trade.get("conditionId") or "" + if not cid: + logger.debug(f"[agg] skipped trade without conditionId: {trade}") + return None + + side = trade.get("side", "") + outcome = trade.get("outcome", "Yes") + try: + price = float(trade.get("price", 0)) + size = float(trade.get("size", 0)) + except (TypeError, ValueError): + logger.debug(f"[agg] skipped trade with bad numerics: {trade}") + return None + + # Compute USD size + if size == 0 and "usdcSize" in trade: + usdc = float(trade.get("usdcSize", 0)) + else: + usdc = size * price + + # Filters + if usdc < self.settings.min_trade_size_usd: + logger.debug( + f"[agg] ${usdc:.0f} < min ${self.settings.min_trade_size_usd:.0f}, skipping" + ) + return None + if not (self.settings.min_price <= price <= self.settings.max_price): + logger.debug( + f"[agg] price {price:.4f} outside [{self.settings.min_price}, {self.settings.max_price}], skipping" + ) + return None + + cred = self.wallet_credibility.get(wallet_addr, 0.5) + if cred < self.settings.min_credibility: + logger.debug( + f"[agg] wallet {wallet_addr[:10]} cred {cred:.2f} < min {self.settings.min_credibility}, skipping" + ) + 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}") + return None + self.wallet_debounce[deb_key] = time.time() + + acc = self.accumulators.get(cid) + 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, + 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]}") + 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) + else: + return None + + logger.debug( + f"[agg] [{cid[:10]}] {wallet_addr[:10]} {side} ${usdc:.0f} @ {price:.4f} | " + f"buy={acc.buy_strength:.2f} ({len(acc.contributors_buy)}w) " + f"sell={acc.sell_strength:.2f} ({len(acc.contributors_sell)}w) | " + f"cred={cred:.2f}" + ) + + self._cleanup_expired() + + 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) + 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) + return None + + def _emit( + self, + cid: str, + side: str, + outcome: str, + price: float, + acc: MarketAccumulator, + ) -> dict: + 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) + total_size_usd = sum(c["size_usd"] for c in contributors) + + kelly = self.sizer.fraction(aggregated, price, side) + suggested = self.sizer.position_usd( + kelly, self.settings.initial_capital_usd + ) + + wallet_contribs = [ + [c["address"], c["size_usd"], c["credibility"]] for c in contributors + ] + + signal = { + "condition_id": cid, + "market_question": acc.market_question, + "side": side, + "outcome": outcome, + "entry_price": price, + "aggregated_strength": aggregated, + "source_wallets": wallet_contribs, + "kelly_fraction": kelly, + "suggested_size_usd": suggested, + "total_signal_size_usd": total_size_usd, + "n_contributors": n, + } + self._signals_emitted += 1 + logger.info( + f"[agg] SIGNAL #{self._signals_emitted}: " + f"{side} {outcome} on {acc.market_question[:60]} @ {price:.4f} | " + f"strength={aggregated:.2f} wallets={n} " + f"kelly={kelly:.3f} size=${suggested:.0f}" + ) + self.accumulators.pop(cid, None) + return signal + + def _cleanup_expired(self) -> None: + now = time.time() + expired = [ + cid + for cid, acc in self.accumulators.items() + if now - acc.last_trade_at > self.settings.consensus_window_seconds + ] + for cid in expired: + acc = self.accumulators.pop(cid, None) + if acc: + logger.debug( + f"[agg] window expired for {cid[:10]} (no consensus reached)" + ) diff --git a/src/services/bayesian.py b/src/services/bayesian.py new file mode 100644 index 0000000..dc200d2 --- /dev/null +++ b/src/services/bayesian.py @@ -0,0 +1,83 @@ +"""Bayesian credibility updater. + +For each wallet, fetch recent closed-positions (realized PnL) and adjust +credibility using a simple posterior update. + +State: + prior_skill: BAYESIAN_PRIOR_SKILL (default 0.5) + posterior: updated based on realized PnL trend + +Update rule (simple exponential smoothing): + if realized_30d > 0: cred += step * (1 - cred) + if realized_30d < 0: cred -= step * cred + +Where step = 0.05 by default (slow update). +""" +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 + +logger = logging.getLogger(__name__) + + +class BayesianUpdater: + """Periodically refresh wallet credibility from realized PnL.""" + + def __init__( + self, + db: CopyTraderDatabase, + aggregator: SignalAggregator, + data: DataAPIClient = None, + ): + 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") + + async def _update_once(self) -> None: + addresses = self.db.get_wallet_addresses() + 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 + 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 + if realized > 0: + new = current + step * (1 - current) + elif realized < 0: + new = current - step * current + else: + new = current + new = max(0.05, min(0.95, new)) + self.aggregator.update_credibility(addr, new) + logger.debug( + f"[bayes] {addr[:10]} realized=${realized:.0f} cred {current:.3f}→{new:.3f}" + ) + logger.info(f"[bayes] credibility refresh complete") + + def stop(self) -> None: + self._running = False diff --git a/src/services/data_api.py b/src/services/data_api.py new file mode 100644 index 0000000..022d129 --- /dev/null +++ b/src/services/data_api.py @@ -0,0 +1,117 @@ +"""Polymarket public API clients (Gamma + Data). All endpoints are public, no auth.""" +import logging +from typing import Any, Dict, List, Optional + +from src.utils.http import get_client + +logger = logging.getLogger(__name__) + +GAMMA_BASE = "https://gamma-api.polymarket.com" +DATA_BASE = "https://data-api.polymarket.com" + + +class DataAPIClient: + """Client for Polymarket Data API (positions, holders, activity).""" + + def __init__(self): + self._client = get_client(timeout=30.0) + + def get_top_holders( + self, condition_id: str, limit: int = 50, timeout: Optional[float] = None, + ) -> List[dict]: + """Fetch top holders by TOTAL_PNL for a market.""" + kwargs: Dict[str, Any] = {} + if timeout is not None: + kwargs["timeout"] = timeout + r = self._client.get( + f"{DATA_BASE}/v1/market-positions", + params={ + "market": condition_id, + "status": "ALL", + "sortBy": "TOTAL_PNL", + "sortDirection": "DESC", + "limit": limit, + }, + **kwargs, + ) + r.raise_for_status() + return r.json() + + def get_positions( + self, address: str, limit: int = 500, timeout: Optional[float] = None, + ) -> List[dict]: + """Fetch all positions for a wallet.""" + kwargs: Dict[str, Any] = {} + if timeout is not None: + kwargs["timeout"] = timeout + r = self._client.get( + f"{DATA_BASE}/positions", + params={"user": address, "limit": limit, "sortBy": "CASHPNL"}, + **kwargs, + ) + r.raise_for_status() + return r.json() + + def get_closed_positions(self, address: str, limit: int = 100) -> List[dict]: + """Closed positions (realized PnL).""" + r = self._client.get( + f"{DATA_BASE}/closed-positions", + params={"user": address, "limit": limit, "sortBy": "REALIZEDPNL"}, + ) + r.raise_for_status() + return r.json() + + 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( + f"{DATA_BASE}/activity", + params={ + "user": address, + "type": "TRADE", + "limit": min(limit, 500), + "sortBy": "TIMESTAMP", + "sortDirection": "DESC", + }, + ) + r.raise_for_status() + return r.json() + + def get_market(self, condition_id: str) -> Optional[dict]: + """Single market metadata from Gamma API.""" + try: + r = self._client.get( + f"{GAMMA_BASE}/markets/{condition_id}", + ) + r.raise_for_status() + return r.json() + except Exception: + return None + + +class GammaAPIClient: + """Client for Gamma API (markets discovery, metadata).""" + + def __init__(self): + self._client = get_client(timeout=30.0) + + def get_active_events_by_volume(self, limit: int = 200) -> List[dict]: + """List active events sorted by 24h volume.""" + r = self._client.get( + f"{GAMMA_BASE}/events", + params={ + "active": "true", + "closed": "false", + "archived": "false", + "order": "volume24hr", + "ascending": "false", + "limit": min(limit, 500), + }, + ) + r.raise_for_status() + return r.json() + + def get_event_by_slug(self, slug: str) -> Optional[dict]: + r = self._client.get(f"{GAMMA_BASE}/events", params={"slug": slug}) + r.raise_for_status() + data = r.json() + return data[0] if data else None diff --git a/src/services/kelly.py b/src/services/kelly.py new file mode 100644 index 0000000..b2fc1a8 --- /dev/null +++ b/src/services/kelly.py @@ -0,0 +1,69 @@ +"""Kelly position sizer with Favorite-Longshot bias correction. + +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 + q = 1 - p + f* = (b * p - q) / b + use = f* * kelly_fraction (default 0.5 → half-Kelly) + +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 math + +from src.config import get_settings + + +class KellySizer: + """Position sizing per the research framework.""" + + def __init__(self): + self.settings = get_settings() + + def win_probability(self, strength: float) -> float: + """Map signal strength [0,1] → win probability. + + strength=0.5 → p=0.55 (baseline) + strength=1.0 → p=0.85 (strong consensus) + strength=0.0 → p=0.50 (coin flip) + """ + strength = max(0.0, min(1.0, strength)) + return 0.50 + 0.35 * strength + + def payoff_ratio(self, price: float, side: str) -> float: + """How much we win vs how much we risk.""" + p = max(0.01, min(0.99, price)) + if side == "BUY": + return (1.0 - p) / p # win=(1-p), risk=p + # SELL: assume we already hold the position at avg price p, hedge at current + return p / (1.0 - p) + + def favorite_longshot_correction(self, price: float, beta: float = 1.5) -> float: + """Smooth penalty for extreme prices. 1.0 at price=0.5, ~0 at extremes.""" + return (1.0 - 2.0 * abs(price - 0.5)) ** beta + + def fraction( + self, + signal_strength: float, + price: float, + side: str, + beta: float = 1.5, + ) -> float: + """Compute Kelly fraction (capped 0..1) for a single signal.""" + p = self.win_probability(signal_strength) + q = 1.0 - p + b = self.payoff_ratio(price, side) + + f_star = max(0.0, (b * p - q) / b) + f_star *= self.favorite_longshot_correction(price, beta) + f_star *= self.settings.kelly_fraction + + return min(f_star, self.settings.max_position_pct) + + def position_usd(self, fraction: float, capital: float) -> float: + """Translate fraction → dollar size, capped by per-trade position limit.""" + size = fraction * capital + max_size = capital * self.settings.max_position_pct + return min(size, max_size) diff --git a/src/services/telegram.py b/src/services/telegram.py new file mode 100644 index 0000000..eb2630b --- /dev/null +++ b/src/services/telegram.py @@ -0,0 +1,65 @@ +"""Telegram bot for trade notifications.""" +import asyncio +import logging +from typing import Optional + +from src.config import get_settings + +logger = logging.getLogger(__name__) + + +class TelegramNotifier: + """Sends trade alerts via Telegram bot.""" + + def __init__(self): + self.settings = get_settings() + self._bot = None + + async def start(self) -> None: + if not self.settings.telegram_enabled: + logger.info("Telegram notifications disabled (TELEGRAM_ENABLED=false)") + return + if not self.settings.telegram_bot_token or not self.settings.telegram_chat_id: + logger.warning("Telegram credentials missing; notifier disabled") + return + try: + from telegram import Bot + self._bot = Bot(token=self.settings.telegram_bot_token) + me = await self._bot.get_me() + logger.info(f"Telegram bot started: @{me.username}") + 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: + if not self._bot: + return + 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" + ) + try: + await self._bot.send_message( + chat_id=self.settings.telegram_chat_id, + text=text, + parse_mode="Markdown", + disable_web_page_preview=True, + ) + except Exception as e: + logger.error(f"Failed to send Telegram signal: {e}") + + async def send_error(self, message: str) -> None: + if not self._bot: + return + try: + await self._bot.send_message( + chat_id=self.settings.telegram_chat_id, + text=f"⚠️ Error: {message[:500]}", + ) + except Exception: + pass diff --git a/src/services/user_stream.py b/src/services/user_stream.py new file mode 100644 index 0000000..cf26199 --- /dev/null +++ b/src/services/user_stream.py @@ -0,0 +1,135 @@ +"""User trade stream (poll-based). Detects new trades from each top wallet. + +Strategy: + - For each wallet, poll /activity?type=TRADE every USER_POLL_INTERVAL_SECONDS + - 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 +""" +import asyncio +import logging +import time +from typing import Awaitable, Callable, Dict, List, Optional, Set + +from src.config import get_settings +from src.db.database import CopyTraderDatabase +from src.services.data_api import DataAPIClient + +logger = logging.getLogger(__name__) + +TradeCallback = Callable[[str, dict], Awaitable[None]] + + +class UserTradeStream: + """Polls each tracked wallet and emits new trades.""" + + def __init__(self, db: CopyTraderDatabase, data: Optional[DataAPIClient] = None): + self.db = db + self.settings = get_settings() + self.data = data or DataAPIClient() + self._last_seen: Dict[str, Optional[str]] = {} + self._running = False + self._poll_count = 0 + self._trades_emitted = 0 + + 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 + logger.info( + f"[stream] starting — poll_interval={self.settings.user_poll_interval_seconds}s, " + f"warmup={self.settings.stream_warmup_seconds}s" + ) + try: + while self._running: + self._poll_count += 1 + addresses = self.db.get_wallet_addresses() + in_warmup = time.time() < warmup_until + if in_warmup and self._poll_count == 1: + logger.info(f"[stream] warming up, will skip baseline trades") + + # Parallel wallet polling (semaphore limits concurrency) + sem = asyncio.Semaphore(self.settings.wallet_pool_concurrency) + poll_sem = self.settings.wallet_pool_concurrency + poll_timeout = self.settings.wallet_pool_request_timeout + + async def poll_one(addr: str) -> None: + async with sem: + try: + trades = await asyncio.wait_for( + asyncio.to_thread( + self.data.get_activity, addr, + self.settings.stream_max_trades_per_wallet, + ), + timeout=poll_timeout, + ) + 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", + ) + + await asyncio.gather(*[poll_one(addr) for addr in addresses]) + + if self._poll_count % 10 == 0: + logger.info( + f"[stream] poll #{self._poll_count} complete " + f"({len(addresses)} wallets, {self._trades_emitted} total trades emitted)" + ) + await asyncio.sleep(self.settings.user_poll_interval_seconds) + except asyncio.CancelledError: + logger.info("[stream] cancelled, shutting down") + + async def _process_trades( + self, + address: str, + trades: List[dict], + on_trade: TradeCallback, + in_warmup: bool, + ) -> None: + last_hash = self._last_seen.get(address) + new_trades: List[dict] = [] + for trade in trades: + tx_hash = self._trade_hash(trade) + if tx_hash is None: + continue + if tx_hash == last_hash: + break + new_trades.append(trade) + + if trades: + first_hash = self._trade_hash(trades[0]) + if first_hash is not None: + self._last_seen[address] = first_hash + + if in_warmup: + logger.info( + f"[stream] warmup: skipped {len(new_trades)} baseline trades for {address[:10]}" + ) + return + + for trade in reversed(new_trades): # process oldest-first + if self.settings.debug_log_api_payloads: + logger.info(f"[stream] NEW TRADE from {address[:10]}: {trade}") + try: + await on_trade(address, trade) + self._trades_emitted += 1 + except Exception as e: + logger.warning( + f"[stream] on_trade callback failed for {address[:10]}: {e}" + ) + + @staticmethod + def _trade_hash(trade: dict) -> Optional[str]: + """Stable unique key per Polymarket trade record.""" + for k in ("transactionHash", "tx_hash", "id", "tradeId"): + v = trade.get(k) + if v: + return str(v) + return None + + def stop(self) -> None: + self._running = False diff --git a/src/services/wallet_pool.py b/src/services/wallet_pool.py new file mode 100644 index 0000000..ed33853 --- /dev/null +++ b/src/services/wallet_pool.py @@ -0,0 +1,217 @@ +"""Top-wallet pool builder. Identifies and maintains the pool of wallets we follow. + +Strategy: + 1. Fetch top events by 24h volume (1 API call) + 2. For each market (parallel, semaphore-limited), fetch top holders + 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 +""" +import asyncio +import logging +from datetime import datetime +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 + +logger = logging.getLogger(__name__) + + +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: + return 0.0 + + pnl_score = min(pnl_30d / 10000.0, 1.0) * 50 + trade_score = min(trades / 100.0, 1.0) * 25 + diversity_score = min(categories / 5.0, 1.0) * 25 + + return pnl_score + trade_score + diversity_score + + +class WalletPoolBuilder: + """Builds and refreshes the target wallet pool.""" + + def __init__(self, db: CopyTraderDatabase): + self.db = db + self.settings = get_settings() + self.gamma = GammaAPIClient() + self.data = DataAPIClient() + + async def build_pool_async(self, max_markets: int = 200) -> List[dict]: + """Async build with parallelism + per-call timeout. + + Returns list of wallet dicts (already filtered by min_pnl/trades/categories). + """ + settings = self.settings + concurrency = settings.wallet_pool_concurrency + timeout = settings.wallet_pool_request_timeout + progress_every = settings.wallet_pool_progress_every + + logger.info( + f"[pool] Building top wallet pool from up to {max_markets} markets " + 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 + ) + 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)" + ) + + if not candidates: + logger.warning("[pool] No candidate wallets found") + return [] + + logger.info(f"[pool] Found {len(candidates)} candidate wallets") + + # Phase 3: parallel position profile fetch + async def fetch_wallet(addr: str) -> dict: + async with semaphore: + try: + positions = await asyncio.to_thread( + self.data.get_positions, addr, 500, timeout * 2 + ) + except Exception as e: + logger.debug(f"[pool] positions failed for {addr[:10]}: {e}") + positions = [] + + 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 + ) + 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") + }) + + score = compute_health_score( + pnl_30d=total_pnl, + total_pnl=total_pnl, + trades=trades, + categories=categories, + ) + return { + "address": addr, + "source": "top_holders", + "pnl_30d_usd": total_pnl, + "pnl_total_usd": total_pnl, + "trades_count": trades, + "categories_count": categories, + "health_score": score, + "credibility": settings.bayesian_prior_skill, + "last_seen_at": datetime.now().isoformat(), + } + + wallet_tasks = [fetch_wallet(addr) for addr in candidates] + wallets: List[dict] = [] + completed = 0 + total = len(wallet_tasks) + + for coro in asyncio.as_completed(wallet_tasks): + wallet = await coro + completed += 1 + 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) + if completed % progress_every == 0 or completed == total: + logger.info( + f"[pool] profiles: scanned {completed}/{total} wallets " + f"({len(wallets)} passed filter so far)" + ) + + wallets.sort(key=lambda w: w["health_score"], reverse=True) + result = wallets[:settings.wallet_pool_size] + logger.info( + f"[pool] Selected top {len(result)} wallets (from {len(wallets)} candidates)" + ) + return result + + def build_pool(self, max_markets: int = 200) -> List[dict]: + """Sync wrapper for CLI usage.""" + return asyncio.run(self.build_pool_async(max_markets)) + + def refresh(self, max_markets: int = 200) -> int: + """Build pool and persist. Returns count. CLI/sync only — call + build_pool_async() directly from async contexts.""" + try: + asyncio.get_running_loop() + logger.warning( + "[pool] refresh() called from async context; " + "use build_pool_async() instead" + ) + return 0 + except RuntimeError: + pass + + pool = asyncio.run(self.build_pool_async(max_markets)) + now = datetime.now().isoformat() + for w in pool: + w.setdefault("added_at", now) + self.db.upsert_wallet_target(w) + return len(pool) diff --git a/src/utils/__init__.py b/src/utils/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/utils/http.py b/src/utils/http.py new file mode 100644 index 0000000..38b30de --- /dev/null +++ b/src/utils/http.py @@ -0,0 +1,24 @@ +"""Shared HTTP client factory with proxy support.""" +import httpx +from src.config.settings import get_settings + + +def get_proxy_config() -> str | None: + settings = get_settings() + proxy = settings.http_proxy.strip() + return proxy if proxy else None + + +def get_client(**kwargs) -> httpx.Client: + proxy = get_proxy_config() + if proxy: + kwargs.setdefault("proxy", proxy) + kwargs.setdefault("timeout", 30.0) + return httpx.Client(**kwargs) + + +def get_async_client(**kwargs) -> httpx.AsyncClient: + proxy = get_proxy_config() + if proxy: + kwargs.setdefault("proxy", proxy) + return httpx.AsyncClient(**kwargs) diff --git a/src/utils/logger.py b/src/utils/logger.py new file mode 100644 index 0000000..94ab9e3 --- /dev/null +++ b/src/utils/logger.py @@ -0,0 +1,73 @@ +"""Logging utilities.""" +import logging +from typing import Optional + +from rich.console import Console +from rich.logging import RichHandler + +from src.config import get_settings + + +def setup_logging(level: Optional[str] = None) -> None: + settings = get_settings() + log_level = level or settings.log_level + + console = Console() + logging.basicConfig( + level=log_level, + format="%(message)s", + datefmt="[%X]", + handlers=[ + RichHandler( + console=console, + rich_tracebacks=True, + show_path=False, + ) + ], + ) + logging.getLogger("httpx").setLevel(logging.WARNING) + logging.getLogger("httpcore").setLevel(logging.WARNING) + logging.getLogger("telegram").setLevel(logging.WARNING) + + +def get_logger(name: str) -> logging.Logger: + return logging.getLogger(name) + + +class BotLogger: + """Custom logger for the copy-trader with formatted output.""" + + def __init__(self): + self.console = Console() + self.logger = logging.getLogger("copy_trader") + + 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]{'='*60}[/bold green]\n" + f"[green]Target Wallets:[/green] {wallet_count}\n" + f"[green]Capital:[/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]{'='*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"[bold magenta]{'='*60}[/bold magenta]\n" + ) + + def info(self, message: str) -> None: + self.console.print(f"[blue]ℹ️[/blue] {message}") + + def error(self, message: str) -> None: + self.console.print(f"[bold red]❌ ERROR:[/bold red] {message}") + + def separator(self) -> None: + self.console.print(f"[dim]{'─'*60}[/dim]") diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/probe_api.py b/tests/probe_api.py new file mode 100644 index 0000000..073047e --- /dev/null +++ b/tests/probe_api.py @@ -0,0 +1,42 @@ +"""Test API directly to check rate limiting.""" +import os +import httpx + +proxy = os.environ.get('HTTP_PROXY', '').strip() or None +client = httpx.Client(timeout=10.0, proxy=proxy) + +print('=== Gamma API test ===') +r = client.get( + 'https://gamma-api.polymarket.com/events', + params={'active': 'true', 'closed': 'false', 'order': 'volume24hr', 'ascending': 'false', 'limit': 3}, +) +print(f'Status={r.status_code} events={len(r.json())}') +if r.json(): + print(f'first title: {r.json()[0].get("title", "?")[:50]}') + +print() +print('=== Data API: top 3 events → 1 market each ===') +events_resp = client.get( + 'https://gamma-api.polymarket.com/events', + params={'active': 'true', 'closed': 'false', 'limit': 3}, +).json() + +for ev in events_resp[:3]: + title = ev.get('title', '?')[:40] + markets = ev.get('markets', []) or [] + if not markets: + continue + cid = markets[0].get('conditionId') + if not cid: + continue + r2 = client.get( + 'https://data-api.polymarket.com/v1/market-positions', + params={'market': cid, 'status': 'ALL', 'sortBy': 'TOTAL_PNL', 'limit': 5}, + ) + try: + data = r2.json() + n = len(data) if isinstance(data, list) else -1 + pos_count = sum(len(d.get('positions', [])) for d in data) if isinstance(data, list) else 0 + print(f' tokens={n} positions={pos_count} | {title}') + except Exception as e: + print(f' parse_err={e} body[:80]={r2.text[:80]}') diff --git a/tests/smoke.py b/tests/smoke.py new file mode 100644 index 0000000..bf996f5 --- /dev/null +++ b/tests/smoke.py @@ -0,0 +1,65 @@ +"""Smoke test for project B — run after build to verify wiring.""" +import asyncio +import os + +os.environ["TELEGRAM_ENABLED"] = "false" +os.environ["LOG_LEVEL"] = "INFO" + +from src.db.database import CopyTraderDatabase +from src.services.kelly import KellySizer +from src.services.aggregator import SignalAggregator + +# 1. Settings +from src.config import get_settings +s = get_settings() +print(f"[ok] settings: capital={s.initial_capital_usd} poll={s.user_poll_interval_seconds}s " + f"min_trade={s.min_trade_size_usd} consensus_wallets={s.consensus_min_wallets}") + +# 2. DB +db = CopyTraderDatabase("data/test_smoke.db") +print(f"[ok] db initialized: {db.get_stats()}") + +# 3. Kelly +sizer = KellySizer() +f = sizer.fraction(signal_strength=0.7, price=0.50, side="BUY") +size = sizer.position_usd(f, 10000) +print(f"[ok] kelly f(s=0.7, p=0.5, BUY)={f:.4f} size={size:.2f}") + +# 4. Aggregator +agg = SignalAggregator(db, sizer=sizer) +agg.wallet_credibility["0xtest1"] = 0.8 +print(f"[ok] aggregator ready") + + +async def test_signal_flow(): + sample = { + "conditionId": "0xtest_market_abc", + "side": "BUY", + "outcome": "Yes", + "price": "0.55", + "size": "1000", + "usdcSize": "550", + } + # First trade — needs debounce + consensus + agg.wallet_credibility["0xtest1"] = 0.8 + sig1 = await agg.on_trade("0xtest1", sample) + print(f"[ok] 1-wallet signal: {sig1} (expected None — needs CONSENSUS_MIN_WALLETS=2)") + + # Second wallet + agg.wallet_credibility["0xtest2"] = 0.7 + sig2 = await agg.on_trade("0xtest2", sample) + if sig2: + print(f"[ok] 2-wallet CONSENSUS REACHED:") + print(f" side={sig2['side']} outcome={sig2['outcome']} " + f"strength={sig2['aggregated_strength']:.3f} " + f"kelly={sig2['kelly_fraction']:.3f} " + f"size_usd={sig2['suggested_size_usd']:.2f} " + f"wallets={sig2['n_contributors']}") + else: + print(f"[warn] 2-wallet did not reach consensus — check threshold") + +asyncio.run(test_signal_flow()) + +# Cleanup test db +os.remove("data/test_smoke.db") +print("\n=== Smoke test complete ===") diff --git a/预测市场.md b/预测市场.md new file mode 100644 index 0000000..99fb2ca --- /dev/null +++ b/预测市场.md @@ -0,0 +1,989 @@ +# 预测市场策略与套利 + +## 预测市场概述 + +预测市场（Prediction Markets）是交易与未来事件结果挂钩的合约的市场，合约价格可解读为市场聚合的预测概率 **[\**1\**]**。著名的平台包括 Polymarket、Iowa Electronic Markets、TradeSports 等。 + +## 主要交易策略 + +### 1. 做市策略（Market-Making） + +做市商同时在买卖两侧挂单，通过赚取买卖价差获利。研究表明，做市商在预测市场中至关重要——他们提供流动性，使连续交易成为可能。在 STOCCER 世界杯预测市场中，做市商参与了 **81%** 的合约交易，贡献了 **85%** 的交易量，且其最终资产价值显著高于普通交易者 **[\**2\**]**。 + +### 2. 信息优势策略 + +顶级交易者通过识别定价错误的合约获利。Polymarket 最新数据显示，**前 1% 的用户掌握了 84% 的总利润**，这些收益几乎全部来自能够"跑赢"市场隐含概率的成熟交易者。而 **70.8% 的用户亏损** **[\**3\**]**。 + +### 3. 跨市场套利 + +不同平台之间或同一平台的不同合约之间可能存在套利机会： + +- **基本组合套利**：当所有合约的卖出价之和低于基本组合价格（所有合约 payoff 之和）时，买入所有合约并卖出基本组合可获利；反之亦然。研究显示，在 STOCCER 市场中大幅套利机会极少且被快速修正（10% 阈值的套利机会仅出现 7 次，平均持续 11 分钟） **[\**2\**]**。 +- **跨平台套利**：不同预测市场对同一事件的价格可能存在差异。例如在足球博彩市场中，通过在不同博彩公司之间组合投注可以产生有限但高利润的套利机会 **[\**4\**]**。 +- **博彩公司与博彩交易所之间的套利**：研究表明，在顶级欧洲足球联赛中，**19.2%** 的比赛可通过在博彩公司和博彩交易所之间组合投注获得确定性正回报 **[\**5\**]**。 + +## 亏损的行为预测因子 + +根据 Polymarket 的数百万用户数据分析 **[\**3\**]**： + +| 因子 | 对亏损概率的影响 | +| ---------------------------- | ---------------------------------------- | +| 极端价格交易（<10¢ 或 >90¢） | 亏损概率增加 3 个百分点 | +| 纯流动性提取者（Taker） | 相比纯做市商，亏损概率高 **36 个百分点** | +| 过度交易 | 亏损概率增加 5 个百分点 | +| 品类集中 | 亏损概率增加 13.6 个百分点 | + +## 关键结论 + +1. **预测市场总体价格有效**：价格为 $p$ 的合约约有 $p$% 的概率正确结算，说明市场整体效率高 **[\**3\**]**。 +2. **长期偏误（Favorite-Longshot Bias）**：低概率事件往往被高估，高概率事件被低估，这与期权市场的"波动率微笑"类似 **[\**1\**]**。 +3. **普通投资者预期回报接近于零**（扣除费用后为负），财富从大量非成熟投资者转移给少数成熟做市商和信息交易者 **[\**3\**]**。 + +基于你提供的 Polymarket API 文档和研究文献，以下是**可实际构建的盈利机器人**方案，按盈利潜力从高到低排列： + +------ + +## 1. 🏆 做市商机器人（最推荐，证据最强） + +**核心策略**：在订单簿两侧挂限价单，赚取买卖价差。 + +**为什么最赚钱**：研究显示，从纯吃单者（Taker）转变为纯做市商（Maker），**亏损概率降低 36 个百分点**——这是所有策略中效果最强的单一因子 **[\**3\**]**。做市商占据了 Polymarket 绝大部分利润。 + +**API 使用**： + +- `GET /book?token_id=...` → 获取订单簿快照 +- `POST /books` → 批量获取多个 token 的订单簿 +- `GET /spread?token_id=...` → 监控价差 +- 鉴权写入端点 → 挂限价单（Maker）和撤单 +- WebSocket → 实时订单簿更新 + +**实现要点**： + +- 计算最优买卖报价（参考当前价差 + 历史波动率） +- 库存管理：控制 Yes/No 双边净敞口 +- 动态调整：高波动时放宽价差，低波动时收紧 +- 关注 **Maker Rebate** 返佣机制（部分市场返还 20-25% 费用） + +------ + +## 2. ⚡ 套利机器人 + +### 2.1 组合套利（Combinatorial Arbitrage） + +同一事件下多个互斥市场的价格之和应 ≈ 1，如果偏离则存在套利。 + +**公式**： + +- 若 `P(Yes) + P(No) > 1` → 卖出 Yes 和 No，买入基本组合（获利 = 总和 - 1） +- 若 `P(Yes) + P(No) < 1` → 买入 Yes 和 No，卖出基本组合（获利 = 1 - 总和） + +**API 使用**： + +- `GET /markets`（Gamma）→ 获取同一 event 下的所有 market +- `GET /price?token_id=...&side=BUY|SELL` → 获取最优买卖价 +- `POST /prices` → 批量获取价格 + +**注意**：研究表明套利机会极少且平均持续仅 **11 分钟**，需要高速自动化 **[\**2\**]**。 + +### 2.2 跨市场/跨平台套利 + +- 监控 Polymarket 和 Kalshi 等不同平台对同一事件的定价差异 +- 在价格低的平台买入，价格高的平台卖出 +- 需考虑跨链转账成本和时间差 + +### 2.3 多结果市场套利 + +对于 `negRisk=true` 的多结果市场，所有结果合约价格之和应 ≤ 1。若 `sum < 1`，买入所有结果可保证正回报。 + +------ + +## 3. 📊 信息优势交易机器人 + +**策略**：识别价格偏离合理概率的合约，低买高卖。 + +**核心数据来源**： + +- `GET /events?active=true&closed=false&order=volume24hr` → 高流动性事件 +- `GET /prices-history?market=...&interval=...` → 历史价格序列 +- `GET /markets` → 市场元数据（到期日、状态等） +- `GET /trades` → 成交历史（识别大额订单） + +**实现方式**： + +- **机器学习模型**：用历史价格、成交量、持仓量等特征预测价格走向 +- **参考信息聚合**：将外部数据（民调、新闻情绪）与市场隐含概率对比 +- **均值回归**：价格短期内大幅偏离历史均线时反向交易 + +**注意**：Polymarket 价格整体有效（校准度接近 45 度线），意味着跑赢市场需要**真正的信息优势**，而非简单策略 **[\**3\**]**。 + +------ + +## 4. 💰 流动性挖矿/返佣机器人 + +**策略**：利用 Polymarket 的 Maker Rebate 机制，挂单赚取返佣。 + +**适用市场**（有费用和返佣的市场）： + +- 加密市场：费率 = 0.25 × 价格²，Maker 返佣 20% +- 体育市场：费率 = 0.0175 × 价格，Maker 返佣 25% + +**API 使用**： + +- `GET /events?tag_id=...&active=true` → 按标签筛选目标市场 +- 鉴权写入端点 → 挂 Maker 限价单 +- WebSocket → 监控成交和返佣 + +**盈利模式**：即使做市本身不赚钱，返佣收入也可覆盖成本并产生正收益。 + +------ + +## 5. 🔍 持仓监控/跟单机器人 + +**策略**：追踪顶级交易者的持仓变化，跟随其交易。 + +**API 使用**： + +- `GET /v1/market-positions?market={conditionId}&sortBy=TOKENS&sortDirection=DESC` → 按持仓量排序的 top 持仓人 +- `GET /positions?user=...` → 特定用户的持仓明细 +- `GET /activity?user=...&type=TRADE` → 用户的交易活动 +- `GET /holders?market={conditionId}` → 市场 top 持仓人 + +**实现方式**： + +- 识别持续盈利的地址（通过 Data API 的 `cashPnl` 字段） +- 监控其开仓/平仓行为 +- 延迟跟随（注意：顶级交易者可能已知晓价格发现） + +------ + +## 6. 📈 事件驱动交易机器人 + +**策略**：在事件临近结算时交易，利用结算前的价格发现和流动性集中。 + +**典型场景**： + +- 选举结果公布前几小时的价格波动 +- 体育比赛进行中的实时价格变动 +- 经济数据发布前后的波动 + +**API 使用**： + +- `GET /events?closed=false&end_date_min={now}&end_date_max={+7d}` → 即将到期的事件 +- WebSocket RTDS → 实时价格和成交数据 +- `GET /midpoint?token_id=...` → 实时中间价 + +------ + +## 📋 各策略对比总结 + +| 策略 | 盈利潜力 | 技术难度 | 所需资本 | 核心风险 | 研究支持 | +| ------------ | -------- | -------- | -------- | ------------------ | ------------------ | +| **做市商** | ⭐⭐⭐⭐⭐ | 高 | 中高 | 库存风险、逆向选择 | 最强 **[\**3\**]** | +| **组合套利** | ⭐⭐⭐ | 中 | 中 | 机会稀缺、速度竞争 | 有 **[\**2\**]** | +| **信息交易** | ⭐⭐⭐⭐ | 高 | 中 | 需要真正信息优势 | 有 **[\**3\**]** | +| **返佣挖矿** | ⭐⭐⭐ | 低 | 中 | 价格不利变动 | 有 **[\**3\**]** | +| **跟单** | ⭐⭐ | 低 | 低 | 延迟成本、被操纵 | 间接 | +| **事件驱动** | ⭐⭐⭐ | 中 | 低 | 波动风险 | 一般 | + +------ + +## ⚠️ 关键风险提示 + +1. **70.8% 用户亏损**，预期收益为负（扣除费用后） **[\**3\**]** +2. 价格整体有效，**系统性跑赢需要真正的信息优势** +3. 极端价格交易（<10¢ 或 >90¢）**增加亏损概率 3 个百分点** +4. 过度交易（高频）**增加亏损概率 5 个百分点** +5. 品类集中（只做体育或只做政治）**增加亏损概率 13.6 个百分点** +6. 约 **25% 的历史成交量可能是刷量交易**（wash trading），需注意数据质量 + +**建议从做市商机器人开始**，这是研究证据最充分、盈利能力最强的策略，且 Polymarket 的 API 和返佣机制为此提供了良好支持。 + +持仓监控/跟单机器人具体讲讲 + +# 持仓监控/跟单机器人 — 完整设计指南 + +## 核心原理 + +跟单机器人的目标是：**识别 Polymarket 上持续盈利的顶级交易者，监控其持仓变化，并自动跟随其交易**。 + +研究显示，Polymarket 上 **前 1% 的用户掌握了 84% 的总利润**，这些收益来自他们识别定价错误合约的能力 **[\**3\**]**。这意味着跟对"聪明钱"可以大幅提高胜率。 + +------ + +## 一、如何识别"聪明钱"地址 + +### 方法 1：按累积 PnL 筛选（最直接） + +使用 Data API 的 `/positions` 端点，遍历用户地址并查看 `cashPnl` 字段： + +``` +GET /positions?user=0x...&sortBy=CASHPNL&sortDirection=DESC +``` + +关键字段： + +- `cashPnl` — 已实现盈亏（美元） +- `percentPnl` — 盈亏百分比 +- `realizedPnl` — 已结算的盈亏 +- `totalBought` — 总买入量 + +**筛选条件建议**： + +| 指标 | 入选门槛 | 原因 | +| ------------------- | ---------------- | --------------------------------------------- | +| `cashPnl` > $10,000 | 排除运气成分 | 研究显示前 1% 用户获得 84% 利润 **[\**3\**]** | +| 交易次数 > 100 | 排除小样本偏差 | 中位数用户仅 24 笔交易 **[\**3\**]** | +| 持仓品种数 > 5 | 排除品类集中风险 | 品类集中增加亏损概率 13.6% **[\**3\**]** | +| 做市交易占比 > 20% | 排除纯吃单者 | 做市降低亏损概率 36 个百分点 **[\**3\**]** | + +### 方法 2：按市场持仓量排序（Top Holders） + +``` +GET /v1/market-positions?market={conditionId}&sortBy=TOKENS&sortDirection=DESC&limit=50 +``` + +返回某市场持仓量最大的用户。研究显示，知情交易者的特征包括： + +- **持仓量更大**（比普通交易者大 58%） **[\**6\**]** +- **交易更活跃**（比普通交易者活跃 39%） **[\**6\**]** +- **经验更丰富**（比普通交易者多 15% 经验） **[\**6\**]** +- **同时做多和做空**（不偏向单边） **[\**6\**]** + +### 方法 3：用 Data API 获取全市场活跃用户 + +``` +GET /activity?type=TRADE&limit=500&sortBy=TOKENS&sortDirection=DESC +``` + +通过成交活动量发现活跃交易者，再追踪其 PnL。 + +------ + +## 二、监控体系架构 + +### 数据流设计 + +``` +┌─────────────────┐ ┌─────────────────┐ ┌─────────────────┐ +│ 定时扫描 │ │ WebSocket 实时 │ │ 信号处理 │ +│ Data API │ ──▶ │ 持仓变化 │ ──▶ │ 风控/决策 │ +│ 识别目标地址 │ │ 监控成交 │ │ 执行交易 │ +└─────────────────┘ └─────────────────┘ └─────────────────┘ +``` + +### 组件 1：目标地址池更新（每日/每小时） + +```python +# 伪代码 +def update_target_addresses(): + # 1. 获取所有活跃市场的 top holders + for market in active_markets: + holders = get(f"/v1/market-positions?market={market.conditionId}&sortBy=TOKENS&limit=50") + for holder in holders: + address_pool[holder.user] += 1 # 累计出现次数 + + # 2. 对候选地址查询 PnL + for addr in address_pool: + positions = get(f"/positions?user={addr}") + if positions.cashPnl > 10000 and len(positions) > 20: + targets.append(addr) + + # 3. 按 PnL 排序，保留前 N 个 + return sorted(targets, key=lambda x: x.cashPnl, reverse=True)[:100] +``` + +### 组件 2：实时持仓监控（WebSocket） + +使用 Polymarket 的 RTDS WebSocket 订阅用户事件： + +``` +wss://ws-subscriptions-clob.polymarket.com/ws/user +``` + +或者轮询 Data API： + +``` +GET /positions?user=0x...&sortBy=CURRENT&sortDirection=DESC +``` + +**关键监控指标**： + +| 指标 | 含义 | 跟单决策 | +| -------------- | ---------------- | -------------------- | +| `size` 变化 | 某合约持仓量变化 | 跟买/跟卖 | +| `avgPrice` | 平均入场价 | 判断是否有利可图 | +| `curPrice` | 当前市场价格 | 判断盈亏状态 | +| `cashPnl` 变化 | 累计盈亏变化 | 评估该地址是否仍有效 | + +### 组件 3：信号生成与执行 + +**跟单规则示例**： + +```python +def should_copy_trade(target_addr, market, side, size, price): + # 1. 风控：检查该地址过去 30 天 PnL 是否为正 + if get_pnl_30d(target_addr) < 0: + return False # 近期表现不佳，不跟 + + # 2. 风控：单笔跟单金额不超过总资金 5% + if size * price > total_capital * 0.05: + return False + + # 3. 风控：不跟极端价格交易（<10¢ 或 >90¢） + if price < 0.10 or price > 0.90: + return False # 极端价格增加亏损概率 3% [3] + + # 4. 延迟跟随：等待 2-5 秒，观察价格是否继续朝同一方向 + price_before = get_midpoint(market) + time.sleep(2) + price_after = get_midpoint(market) + if (side == "BUY" and price_after < price_before) or \ + (side == "SELL" and price_after > price_before): + return False # 价格反转，不跟 + + return True +``` + +------ + +## 三、跟单策略的三种变体 + +### 方案 A：精确跟单（复制每笔交易） + +**优点**：完全复制目标地址的持仓 +**缺点**：延迟导致滑点，容易被反向狙击 + +**技术实现**： + +- 通过 Data API `/activity` 轮询或 WebSocket 实时获取目标地址的每笔交易 +- 解析 `type=TRADE`、`side=BUY/SELL`、`asset_id`、`size`、`price` +- 以相同方向、相同比例执行 + +### 方案 B：持仓对齐（周期性同步） + +**优点**：成本更低，延迟容忍度高 +**缺点**：无法捕捉短线机会 + +**技术实现**： + +```python +def sync_positions(): + for target in target_addresses: + target_positions = get(f"/positions?user={target}") + my_positions = get(f"/positions?user={MY_ADDRESS}") + + # 计算每个市场的持仓差异 + for market in target_positions: + target_size = target_positions[market].size + my_size = my_positions.get(market, 0) + gap = target_size * COPY_RATIO - my_size + + if abs(gap) > MIN_TRADE_SIZE: + execute_trade(market, gap) +``` + +### 方案 C：信号聚合（多地址共识） + +**优点**：减少单一地址被操纵或失效的风险 +**缺点**：信号延迟更大 + +**技术实现**： + +- 监控 N 个顶级交易者（建议 N ≥ 10） +- 当 ≥ 70% 的地址在同一方向交易同一市场时，才执行 +- 加权：按 `cashPnl` 或 `percentPnl` 加权 + +------ + +## 四、风控系统（最关键） + +### 风险规则矩阵 + +| 规则 | 具体实现 | 研究依据 | +| ------------------ | ---------------------------------- | ------------------------------------------ | +| **最大持仓集中度** | 单一市场不超过总资金 20% | 品类集中增加亏损概率 13.6% **[\**3\**]** | +| **禁止极端价格** | 不跟价格 < 10¢ 或 > 90¢ 的交易 | 极端价格交易增加亏损概率 3% **[\**3\**]** | +| **做市商偏好** | 优先跟单做市商（Maker 占比 > 30%） | 做市降低亏损概率 36 个百分点 **[\**3\**]** | +| **交易频率限制** | 每小时最多 10 笔 | 过度交易增加亏损概率 5% **[\**3\**]** | +| **目标地址淘汰** | 连续 7 天 PnL 为负则淘汰 | 技能持续性检验 | +| **延迟执行** | 收到信号后等待 2-10 秒 | 防止被反向狙击 | +| **最大跟单金额** | 单笔不超过总资金 5% | 资金管理基本规则 | + +### 目标地址健康度评分 + +```python +def health_score(address): + score = 0 + positions = get(f"/positions?user={address}") + + # 1. PnL 正向（权重 40%） + cash_pnl = sum(p.cashPnl for p in positions) + score += min(cash_pnl / 10000, 1) * 40 + + # 2. 做市商占比（权重 30%） + maker_ratio = get_maker_ratio(address) + score += maker_ratio * 30 # 越高越好 + + # 3. 品类分散度（权重 20%） + category_hhi = get_category_hhi(address) + score += (1 - category_hhi) * 20 # HHI 越低越好 + + # 4. 近期表现（权重 10%） + recent_pnl = get_pnl_30d(address) + score += max(min(recent_pnl / 1000, 1), 0) * 10 + + return score +``` + +------ + +## 五、完整实现流程 + +### 步骤 1：初始化目标地址池 + +```python +# 每天运行一次 +candidates = [] + +# 方法 A：从 top holders 中挖掘 +for market in top_volume_markets(limit=200): + holders = get(f"/v1/market-positions?market={market.conditionId}&sortBy=TOKENS&limit=50") + for h in holders: + candidates.append(h.user) + +# 方法 B：从历史交易中挖掘活跃地址 +for trade in get("/trades?limit=10000"): + candidates.append(trade.maker_address) + candidates.append(trade.owner) + +# 去重并筛选 +candidates = list(set(candidates)) +for addr in candidates: + info = get(f"/positions?user={addr}") + if info.cashPnl > 5000 and len(info) > 10: + target_pool.add(addr) +``` + +### 步骤 2：实时监控循环 + +```python +while True: + for target in target_pool: + # 每隔 30 秒拉取一次持仓 + positions = get(f"/positions?user={target}") + for pos in positions: + if pos.conditionId not in my_positions: + # 新开仓，执行跟单 + if should_copy_trade(target, pos): + execute_trade(pos.conditionId, pos.side, + pos.size * COPY_RATIO) + else: + # 持仓变化 + diff = pos.size - my_positions[pos.conditionId] + if abs(diff) > MIN_DELTA: + adjust_position(pos.conditionId, diff * COPY_RATIO) + + # 更新目标地址评分 + for target in target_pool: + target.score = health_score(target) + + # 淘汰低分地址 + target_pool = [t for t in target_pool if t.score > 50] + + time.sleep(30) # 轮询间隔 +``` + +### 步骤 3：执行交易（CLOB API） + +```python +def execute_trade(condition_id, side, size): + # 获取 token_id + market = get(f"/markets/{condition_id}") + token_id = market.token_id if side == "BUY" else market.no_token_id + + # 获取最优价格 + price = get(f"/price?token_id={token_id}&side=SELL" if side == "BUY" else "BUY") + + # 挂限价单（Maker），获取返佣 + order = { + "token_id": token_id, + "side": "BUY" if side == "BUY" else "SELL", + "size": str(size), + "price": str(price), + "signature": sign_message(...) + } + response = post("/order", order, headers=auth_headers) +``` + +------ + +## 六、关键注意事项 + +### 1. 延迟问题 + +跟单天然存在延迟。研究表明，知情交易者的信息优势可能在几分钟内被市场消化 **[\**6\**]**。建议： + +- **低流动性市场**：延迟 5-10 秒，观察价格是否稳定 +- **高流动性市场**：延迟 2-3 秒，快速执行 +- **使用 WebSocket** 而非轮询，减少延迟 + +### 2. 目标地址失效 + +研究显示，顶级交易者的技能可能随时间衰减 **[\**6\**]**。建议： + +- 持续监控目标地址的近期 PnL +- 每周重新评估目标地址池 +- 保持至少 20-50 个目标地址以分散风险 + +### 3. 被反向操纵的风险 + +公开的地址可能被恶意操纵。防范措施： + +- 不跟单仅 1-2 个地址，至少 10 个以上 +- 使用信号聚合（多地址共识） +- 加入价格验证（检查交易是否合理） + +### 4. 费用考虑 + +- 部分市场有 Taker 费用（加密市场 0.25 × 价格²，体育市场 0.0175 × 价格） +- 尽量使用 Maker 限价单执行，可获得 20-25% 返佣 +- 高频跟单可能因费用侵蚀利润 + +------ + +## 七、预期收益与现实 + +**现实数据**： + +- 70.8% 的用户亏损 **[\**3\**]** +- 预期收益为 0（扣除费用后为负） **[\**3\**]** +- 前 1% 用户获得 84% 利润 **[\**3\**]** + +**跟单机器人的优势**： + +- 避免了"极端价格交易"（63% 的普通用户犯此错误） **[\**3\**]** +- 避免了"过度交易"（增加亏损概率 5%） **[\**3\**]** +- 避免了"品类集中"（增加亏损概率 13.6%） **[\**3\**]** + +**跟单机器人的局限**： + +- 无法完全消除延迟滑点 +- 顶级交易者的交易量可能很大，小资金跟单效果有限 +- 无法复制做市商策略（做市需要实时双边报价，难以跟单） + +**建议组合策略**：跟单 + 做市商混合。将 70% 资金用于做市商策略（被证明最赚钱），30% 用于跟单（捕捉信息优势交易）。 + +你懂这么多，能不能使用数学/统计学，可量化行为学等等角度优化跟单策略吗 + +# 跟单策略的数学与统计优化框架 + +这是一个很好的问题。跟单本质上是**信号跟踪问题**，可以从多个数学角度系统性地优化。下面我从五个层面构建优化框架。 + +------ + +## 一、信号提取：从持仓数据中提取知情交易信号 + +### 1.1 持仓变化信号 vs 交易事件信号 + +跟单面临一个核心问题：**顶级交易者可能是做市商（Maker）而非吃单者（Taker）**。做市商的持仓变化是**被动产生**的，而非主动信号。 + +**统计分解**： + +设目标地址 $u$ 在时间 $t$ 对合约 $i$ 的持仓量为 $Q_{u,i,t}$。持仓变化可分解为： + +$$\Delta Q_{u,i,t} = \underbrace{\Delta Q_{u,i,t}^{\text{主动}}}*{\text{信号}} + \underbrace{\Delta Q*{u,i,t}^{\text{做市}}}_{\text{噪音}}$$ + +其中做市部分可通过分析交易对手方向来识别——如果该地址同时存在同一合约的 Yes 和 No 持仓，则更可能在做市。 + +**优化方案**：使用 Data API 的 `/activity` 端点，筛选 `type=TRADE`，仅关注目标地址**主动开仓**的交易（非被动成交）。主动开仓的特征是： + +- `trader_side` 与持仓方向一致 +- 交易后该方向的持仓占比显著增加 + +### 1.2 持仓变化率与持仓水平信号 + +传统跟单只关注 $\Delta Q$（持仓变化），但更优的信号是**持仓偏离历史均值的程度**： + +定义标准化持仓偏离度： + +$$Z_{u,i,t} = \frac{Q_{u,i,t} - \mu_{u,i}}{\sigma_{u,i}}$$ + +其中 $\mu_{u,i}$ 和 $\sigma_{u,i}$ 是该地址在此合约上的历史持仓均值和标准差。 + +**决策规则**：只有当 $|Z_{u,i,t}| > \theta$（阈值，建议 1.5-2.0）时才跟单。这过滤了做市导致的微小持仓波动。 + +------ + +## 二、信号聚合：多地址加权共识模型 + +### 2.1 贝叶斯加权聚合 + +假设有 $K$ 个目标地址，每个地址 $k$ 对合约 $i$ 的净方向信号为 $s_{k,i} \in {-1, 0, +1}$（卖、中性、买），地址的"信噪比"为 $\alpha_k$（可通过历史 PnL 和预测准确度估计）。 + +贝叶斯聚合信号： + +$$S_i = \frac{\sum_{k=1}^{K} w_k \cdot s_{k,i}}{\sum_{k=1}^{K} w_k}$$ + +其中权重 $w_k$ 由多个维度决定： + +$$w_k = \underbrace{\frac{1}{1 + e^{-\gamma \cdot \text{PnL}*k}}}*{\text{PnL 权重}} \times \underbrace{\frac{1}{1 + \lambda \cdot \text{延迟}*k}}*{\text{延迟惩罚}} \times \underbrace{\min\left(\frac{N_k}{N_0}, 1\right)}_{\text{样本量校正}}$$ + +- $\text{PnL}_k$：该地址累计盈亏（美元） +- $\text{延迟}_k$：该地址信号到我们的平均延迟（秒） +- $N_k$：该地址交易次数 +- $\gamma, \lambda, N_0$：可调超参数 + +**决策规则**： + +- $S_i > \tau$ → 跟买 +- $S_i < -\tau$ → 跟卖 +- $|S_i| \leq \tau$ → 不行动 + +**阈值 $\tau$ 优化**：通过历史回测最小化夏普比率或最大化卡玛比率。 + +### 2.2 隐马尔可夫模型（HMM）信号提取 + +顶级交易者的交易行为可能呈现**状态切换**（如"积极建仓期"→"持仓期"→"平仓期"）。 + +定义可观测变量 $O_t = {\Delta Q_{u,t}, \text{成交频率}, \text{方向一致性}}$，隐含状态 $Z_t \in {\text{建仓}, \text{持有}, \text{减仓}, \text{观望}}$。 + +**参数估计**：用 Baum-Welch 算法从历史数据学习转移矩阵 $A$ 和发射概率 $B$。 + +**实时滤波**：用前向算法计算 $P(Z_t = \text{建仓} | O_{1:t})$，当概率 > 0.7 时触发跟单。 + +------ + +## 三、执行优化：最优跟单规模和时机 + +### 3.1 凯利公式优化仓位 + +假设信号 $S_i$ 对应的预期收益率为 $r$，胜率为 $p$，则**凯利最优仓位**为： + +$$f^* = \frac{p \cdot r - (1-p) \cdot 1}{r} = \frac{p \cdot (1 + r) - 1}{r}$$ + +在 Polymarket 环境中，由于价格 $P$ 可视为概率，预期收益可近似为： + +$$r = \frac{P_{\text{结算}} - P_{\text{入场}}}{P_{\text{入场}}}$$ + +但由于我们不知道真实结算概率，只能用**信号强度 $S_i$ 来近似 $p$**： + +$$\hat{p} = \frac{1}{1 + e^{-\beta S_i}}$$ + +其中 $\beta$ 通过历史数据校准。 + +**实际仓位**（安全凯利）： + +$$f = \min\left(\frac{f^*}{2}, f_{\max}\right)$$ + +### 3.2 延迟优化：最优等待时间 + +延迟是一把双刃剑——等待越久信号越可靠，但执行价格越差。 + +定义**延迟-价格影响函数**：通过历史数据拟合目标地址交易后 $t$ 秒的价格变化： + +$$\Delta P(t) = \mathbb{E}[P_{\text{post}}(t) - P_{\text{pre}}]$$ + +**最优等待时间**： + +$$t^* = \arg\max_t \left[ \underbrace{\Delta P(t)}*{\text{价格变化}} - \underbrace{\kappa \cdot \sigma \cdot \sqrt{t}}*{\text{价格风险}} \right]$$ + +其中 $\kappa$ 是风险厌恶系数，$\sigma$ 是价格波动率。 + +**经验值**：高流动性市场（如 NFL 比赛）$t^* \approx 2-5$ 秒；低流动性市场 $t^* \approx 10-30$ 秒。 + +### 3.3 最优跟单比率 + +并非所有信号都值得 1:1 跟单。定义**信号质量评分**： + +$$\text{SignalScore}*{u,i,t} = \underbrace{|Z*{u,i,t}|}_{\text{偏离度}} \times \underbrace{\text{MakerRatio}^{-1}*u}*{\text{主动交易概率}} \times \underbrace{\text{Recency}*u}*{\text{近期表现}}$$ + +跟单比率： + +$$\lambda_{u,i,t} = \min\left( \frac{\text{SignalScore}*{u,i,t}}{\text{SignalScore}*{\max}}, 1 \right)$$ + +执行规模 = $\lambda \times \Delta Q_u \times \text{资金比例因子}$ + +------ + +## 四、行为金融学：识别并规避认知偏差 + +### 4.1 Favorite-Longshot Bias 校正 + +研究显示，预测市场存在**偏好-冷门偏误**：低概率事件被高估，高概率事件被低估 **[\**1\**]**。这意味着： + +- **跟买低概率合约**（价格 < 20¢）的预期收益为负 +- **跟卖高概率合约**（价格 > 80¢）的预期收益也为负 + +**校正模型**：对信号方向施加价格惩罚因子： + +$$\text{AdjustedSignal}*{i} = S_i \times \underbrace{\left(1 - 2 \cdot |P_i - 0.5|\right)^{\beta}}*{\text{偏误校正因子}}$$ + +其中 $\beta > 0$ 控制校正强度。当 $P_i = 0.5$ 时因子最大（1.0），当 $P_i \to 0$ 或 $1$ 时因子趋近于 0。 + +### 4.2 过度自信校正 + +研究显示，63% 的普通用户交易极端价格合约 **[\**3\**]**。顶级交易者也难免过度自信。 + +**过度自信检测**：统计目标地址在**连续亏损后是否加大仓位**（"翻本效应"）。定义： + +$$\text{OverconfidenceScore}*u = \mathbb{E}\left[\frac{\Delta Q*{u,t+1}}{\text{std}(\Delta Q_u)} \Bigg| \text{PnL}_{u,t} < 0\right]$$ + +若该分数 > 1.5，说明该地址有显著翻本倾向，其信号应降低权重。 + +### 4.3 处置效应（Disposition Effect）识别 + +投资者倾向于**过早卖出盈利头寸**、**持有亏损头寸过久**。 + +**处置效应检测**：统计目标地址在盈利状态下的平仓概率 vs 亏损状态下的平仓概率： + +$$\text{DispositionRatio}_u = \frac{P(\text{卖出} | \text{盈利})}{P(\text{卖出} | \text{亏损})}$$ + +若该比率 > 1.5，说明该地址存在显著处置效应。**优化策略**：当该地址卖出盈利头寸时，我们**不跟卖**（因为可能是过早卖出）；当该地址卖出亏损头寸时，我们**更积极地跟卖**（因为克服了持有偏见）。 + +------ + +## 五、完整数学优化框架 + +### 5.1 信号-执行-风控三阶段模型 + +#### 阶段 1：信号生成 + +$$\hat{S}*{i,t} = \sum*{k=1}^{K} \underbrace{\frac{e^{\eta \cdot \text{Score}*{k,t}}}{\sum*{j} e^{\eta \cdot \text{Score}*{j,t}}}}*{\text{Softmax 权重}} \cdot \underbrace{\tanh\left(\alpha \cdot \frac{\Delta Q_{k,i,t}}{\sigma_{k,i} + \epsilon}\right)}*{\text{持仓变化信号}} + \underbrace{\beta \cdot \text{PriceDeviation}*{i,t}}_{\text{价格偏差信号}}$$ + +#### 阶段 2：执行决策 + +$$f_{i,t}^* = \underbrace{\frac{\hat{S}*{i,t} \cdot \text{Edge}*{i,t} - (1 - \text{Edge}*{i,t})}{\text{Edge}*{i,t}}}_{\text{凯利公式}} \times \underbrace{\left(1 - \frac{\text{当前持仓}*i}{\text{最大持仓}}\right)}*{\text{分散化惩罚}}$$ + +#### 阶段 3：风控约束 + +$$\text{最终仓位} = \min\left(f^* \times \text{资本}, \text{仓位上限}\right) \times \mathbb{1}{\text{HHI} < 0.3} \times \mathbb{1}{\text{最近信号数} > 3}$$ + +其中 HHI（赫芬达尔指数）= $\sum_i (\text{仓位}_i / \text{总仓位})^2$，约束集中度。 + +### 5.2 超参数优化框架 + +可优化的超参数向量 $\Theta = {\tau, \beta, \eta, \alpha, \kappa, f_{\max}, \text{延迟}, N_0}$。 + +**优化目标**： + +$$\Theta^* = \arg\max_{\Theta} \left[ \underbrace{\text{夏普比率}}*{\text{风险调整收益}} - \lambda_1 \cdot \underbrace{\text{最大回撤}}*{\text{下行风险}} - \lambda_2 \cdot \underbrace{\text{换手率}}_{\text{交易成本}} \right]$$ + +**优化方法**：使用**贝叶斯优化**（因参数空间小、目标函数评估成本高）或**网格搜索**（如果计算资源充足）。 + +### 5.3 在线学习：贝叶斯更新地址权重 + +随着新数据到来，持续更新每个目标地址的**后验可信度**： + +$$P(\text{地址}*k \text{有效} | D*{1:t}) = \frac{P(D_{1:t} | \text{有效}) \cdot P(\text{有效})}{P(D_{1:t})}$$ + +假设收益服从正态分布： + +$$P(D_{1:t} | \text{有效}) = \prod_{s=1}^{t} \frac{1}{\sqrt{2\pi\sigma_k^2}} \exp\left(-\frac{(\text{PnL}_{k,s} - \mu_k)^2}{2\sigma_k^2}\right)$$ + +当后验概率低于阈值（如 0.1）时，将该地址从池中移除。 + +------ + +## 六、实战代码框架 + +```python +import numpy as np +from scipy.stats import norm +from hmmlearn import hmm # 隐马尔可夫模型 + +class MathCopyTrader: + def __init__(self, capital=10000, max_targets=50): + self.capital = capital + self.max_targets = max_targets + self.targets = {} # address -> {positions, pnl, maker_ratio, ...} + self.params = { + 'tau': 0.3, # 信号阈值 + 'beta': 1.5, # Favorite-Longshot 校正强度 + 'eta': 2.0, # Softmax 温度参数 + 'kappa': 0.5, # 风险厌恶 + 'delay': 5, # 等待秒数 + 'f_max': 0.2, # 单笔最大仓位 + 'hhi_max': 0.3, # 集中度上限 + } + + def z_score_position(self, addr, contract_id, Q): + """标准化持仓偏离度""" + history = self.targets[addr]['history'][contract_id] + mu = np.mean(history) + sigma = np.std(history) + 1e-8 + return (Q - mu) / sigma + + def favorite_longshot_correction(self, price, signal): + """偏好-冷门偏误校正""" + factor = (1 - 2 * abs(price - 0.5)) ** self.params['beta'] + return signal * factor + + def softmax_weight(self, scores): + """Softmax 归一化权重""" + scores = np.array(scores) + exp_s = np.exp(self.params['eta'] * scores) + return exp_s / exp_s.sum() + + def aggregate_signal(self, contract_id, price): + """多地址信号聚合""" + signals = [] + weights = [] + + for addr, data in self.targets.items(): + # 持仓变化信号 + pos = data['positions'].get(contract_id) + if pos: + z = self.z_score_position(addr, contract_id, pos['size']) + + # 主动交易概率（1 - maker_ratio） + active_prob = 1 - data.get('maker_ratio', 0.5) + + # 近期表现 + recency = data.get('pnl_30d', 0) / 1000 + + signal = np.tanh(z * active_prob) + weight = data.get('health_score', 50) * (1 + recency) + + signals.append(signal) + weights.append(max(weight, 0)) + + if not signals: + return 0 + + # 加权聚合 + w = self.softmax_weight(weights) + aggregated = np.dot(w, signals) + + # Favorite-Longshot 校正 + adjusted = self.favorite_longshot_correction(price, aggregated) + + return adjusted + + def kelly_fraction(self, signal, price, target_price=1.0): + """凯利最优仓位""" + # 用信号强度估计胜率 + p = 1 / (1 + np.exp(-self.params['kappa'] * signal)) + p = np.clip(p, 0.01, 0.99) + + # 赔率 + b = (target_price - price) / price if signal > 0 else price / (1 - price) + + # 凯利公式 + f_star = (p * b - (1 - p)) / b + return np.clip(f_star, 0, self.params['f_max']) / 2 # 半凯利 + + def should_execute(self, signal, current_positions, contract_id): + """风控检查""" + # 1. 信号强度检查 + if abs(signal) < self.params['tau']: + return False + + # 2. 集中度检查 + total_pos = sum(abs(p) for p in current_positions.values()) + if total_pos > 0: + new_pos_ratio = abs(signal) / total_pos + hhi = sum((p / total_pos)**2 for p in current_positions.values()) + if hhi > self.params['hhi_max'] and new_pos_ratio > 0.1: + return False + + # 3. 信号一致性检查（过去3个信号方向是否一致） + recent_signals = [s for s in self.signal_history.get(contract_id, [])[-3:]] + if len(recent_signals) >= 2: + if sum(1 for s in recent_signals if np.sign(s) == np.sign(signal)) < 2: + return False # 信号不一致，不跟 + + return True + + def execute_trade(self, contract_id, signal, price): + """执行交易""" + if not self.should_execute(signal, self.current_positions, contract_id): + return + + # 凯利仓位 + f = self.kelly_fraction(signal, price) + amount = f * self.capital + + # 延迟执行 + time.sleep(self.params['delay']) + + # 检查延迟后的价格是否仍有利 + new_price = get_midpoint(contract_id) + if signal > 0 and new_price > price * 1.02: # 价格已上涨2% + return # 滑点太大，放弃 + if signal < 0 and new_price < price * 0.98: # 价格已下跌2% + return + + # 执行 + side = "BUY" if signal > 0 else "SELL" + place_order(contract_id, side, amount, new_price) + + def update_target_beliefs(self): + """贝叶斯更新地址可信度（在线学习）""" + for addr, data in self.targets.items(): + # 获取近期 PnL 序列 + pnls = data.get('daily_pnl', [])[-30:] + if len(pnls) < 10: + continue + + mu = np.mean(pnls) + sigma = np.std(pnls) + 1e-8 + + # 似然：观察到的 PnL 来自"有效"分布的概率 + likelihood = norm.pdf(mu, loc=0.01, scale=0.05) # 有效分布均值 > 0 + + # 后验 + prior = data.get('prior_skill', 0.5) + posterior = (likelihood * prior) / (likelihood * prior + (1 - likelihood) * (1 - prior)) + + data['prior_skill'] = posterior + + # 淘汰低分地址 + if posterior < 0.1 and len(self.targets) > 10: + del self.targets[addr] +``` + +------ + +## 七、回测与验证框架 + +### 7.1 回测设计 + +**数据需求**： + +- 至少 3-6 个月的完整交易历史 +- 包含所有地址的持仓快照（分钟级） +- 价格数据（分钟级 tick data） + +**评估指标**： + +| 指标 | 公式 | 目标值 | +| -------- | -------------------------------------------- | ------ | +| 夏普比率 | $\frac{\mu_R - r_f}{\sigma_R}$ | > 1.0 | +| 卡玛比率 | $\frac{\mu_R - r_f}{\text{MaxDrawdown}}$ | > 0.5 | +| 胜率 | $\frac{\text{盈利交易}}{\text{总交易}}$ | > 55% | +| 盈亏比 | $\frac{\text{平均盈利}}{\text{平均亏损}}$ | > 1.5 | +| 信息比率 | $\frac{\alpha}{\sigma_\epsilon}$（相对基准） | > 0.5 | + +### 7.2 反事实分析 + +对于每个未被跟单的信号，计算**放弃收益**（如果跟了会赚/亏多少）。这用于评估风控规则是否过于保守。 + +### 7.3 蒙特卡洛模拟 + +对超参数 $\Theta$ 进行随机采样（1000+ 次），观察目标函数分布，确保优化结果不是过拟合的产物。 + +------ + +## 总结：数学优化的边际收益估计 + +| 优化层 | 预期夏普提升 | 难度 | 优先级 | +| ------------------------- | ------------ | ---- | ------ | +| 信号聚合（Softmax + HMM） | +0.3-0.5 | 中 | ⭐⭐⭐⭐⭐ | +| 凯利仓位管理 | +0.2-0.4 | 低 | ⭐⭐⭐⭐⭐ | +| Favorite-Longshot 校正 | +0.1-0.3 | 低 | ⭐⭐⭐⭐ | +| 延迟优化 | +0.1-0.2 | 中 | ⭐⭐⭐ | +| 贝叶斯地址淘汰 | +0.1-0.2 | 中 | ⭐⭐⭐ | +| 处置效应检测 | +0.05-0.15 | 高 | ⭐⭐ | +| HMM 状态识别 | +0.05-0.15 | 高 | ⭐⭐ | + +**建议优先实现**：信号聚合 + 凯利仓位 + Favorite-Longshot 校正，这三项性价比最高。 \ No newline at end of file