refactor: remove legacy trading engine, streamline project to weather-only bot

This commit is contained in:
AmandaloveYang
2026-02-18 09:56:50 +08:00
parent d533bd5b21
commit 30f86eab89
29 changed files with 25 additions and 3644 deletions
-74
View File
@@ -1,74 +0,0 @@
# Polymarket Weather Market Discovery Technical Documentation
> ⚠️ **Current Status: Suspended**
> The automated market discovery and monitoring engine described here has been commented out in `run.py`. The system currently operates in "passive query mode" — weather analysis is only triggered by the `/city` command.
This document explains the technical implementation of how PolyWeather identifies and tracks weather markets on Polymarket.
## 1. Data Sources
We bypass high-level SDKs and interact directly with the **Polymarket Gamma API**, which is the primary metadata layer for Discovery.
- **Base URL:** `https://gamma-api.polymarket.com`
- **Endpoint:** `/markets`
## 2. Discovery Strategy
The system uses a multi-layered search approach to ensure no city segments are missed.
### 2.1 Keyword Triple-Search
Instead of one query, we execute three concurrent search patterns:
1. `"highest temperature"`: Targets the primary question text.
2. `"temperature in"`: Broad search for regional markets.
3. `"daily weather"`: Fallback for markets with different naming conventions.
### 2.2 Prioritization
We apply specific sorting to find the **latest** available contracts (e.g., February 9th, 2026):
- `order=id` & `ascending=false`: Scans the newest created markets first.
- `active=true` & `closed=false`: Filters out resolved or expired contracts.
## 3. Filtering & Parsing Logic
Since Polymarket hosts thousands of events, we apply a strict "Weather Filter" in the code:
### 3.1 Text Validation
We inspect both the `question` and the `slug`:
- **Pattern Match:** Must contain `"highest temperature in"` or `"highest-temperature-in"`.
- **Exclusion:** (Implicitly handled by keyword search) filtered from sports or politics.
### 3.2 Negative Risk Market Handling
Weather markets on Polymarket are often structured as **Negative Risk** groups (where multiple outcomes like "70°F or higher" and "68-69°F" belong to one event).
**Technical Challenge:** In the API's list view, the `activeTokenId` field is often `null` for these complex markets.
**Our Solution:**
1. Check `clobTokenIds`.
2. If it's a JSON string (common in Gamma), parse it into a Python list.
3. If `activeTokenId` is missing, we treat the first token ID in the list as the **"YES" Token**.
4. This allows us to fetch the real-time orderbook/price even for markets that haven't fully "activated" in the front-end metadata.
## 4. Market Data Structure
Every market found is normalized into this structure for the Decision Engine:
- `condition_id`: The UMA condition ID for resolution.
- `active_token_id`: The specific ERC1155 token ID we want to buy/monitor.
- `group_id`: The `negRiskMarketID`, which allows the bot to understand that specific temperature ranges (e.g., 70°F vs 72°F) are related to the same city.
- `slug`: Used for generating direct dashboard links.
## 5. Frequency & Caching
- **Discovery Frequency:** The system rescans for new cities/dates every **5 minutes**.
- **Caching:** Found markets are stored in an internal memory cache (`_weather_markets_cache`) to reduce API pressure and avoid rate limits.
---
_Created on: 2026-02-07_
_PolyWeather System Documentation_
-74
View File
@@ -1,74 +0,0 @@
# Polymarket 天气市场搜寻技术文档
> ⚠️ **当前状态:功能休眠**
> 本中提到的自动搜寻与监控引擎目前已在 `run.py` 中被注释。当前系统工作于“被动查询模式”,仅在用户输入 `/city` 指令时提供天气分析。
本文档详细说明了 PolyWeather 如何在 Polymarket 上自动识别、筛选并跟踪天气相关市场的技术实现逻辑。
## 1. 数据来源
我们跳过了复杂的官方 SDK,直接与 **Polymarket Gamma API** 交互。这是 Polymarket 的官方元数据层,负责所有市场的发现与展示。
- **Base URL:** `https://gamma-api.polymarket.com`
- **Endpoint:** `/markets`
## 2. 搜寻策略
由于 Polymarket 同时挂载数千个预测市场,系统采用多层搜索方案以确保不会遗漏任何城市的分段合约。
### 2.1 关键词三重搜索
程序并非只搜索一个词,而是并发执行三个搜索模式:
1. `"highest temperature"`: 匹配大多数天气问题的核心描述。
2. `"temperature in"`: 针对特定地区市场的宽泛搜索。
3. `"daily weather"`: 针对某些命名不规范市场的兜底搜索。
### 2.3 优先级与排序
为了确保能搜到**最新**发布的合约(例如 2026年2月9日 的市场),我们应用了特定的 API 排序参数:
- `order=id` & `ascending=false`: 优先扫描最新创建的市场 ID。
- `active=true` & `closed=false`: 过滤掉已结算或已关闭的无效合约。
## 3. 过滤与解析逻辑
系统在获取 API 返回的列表后,会进行二次深度筛选:
### 3.1 文本校验
检查市场的 `question`(问题描述)和 `slug`URL 路径):
- **模式匹配:** 必须包含 `"highest temperature in"``"highest-temperature-in"`
- **城市提取:** 逻辑会自动识别问题中的城市名(如 芝加哥、伦敦 等)。
### 3.2 负风险(Negative Risk)市场处理
Polymarket 的天气市场通常以 **Negative Risk** 分组形式存在(一个事件下包含多个互斥的区间,如“70°F以上”和“68-69°F”)。
**技术挑战:** 在 API 的列表视图中,这类市场的 `activeTokenId` 字段经常返回 `null`
**我们的解决方案:**
1. 检查 `clobTokenIds` 字段。
2. 如果该字段是 JSON 字符串(Gamma API 的常见返回格式),则将其解析为 Python 列表。
3. 如果 `activeTokenId` 缺失,我们将列表中的第一个 Token ID 视为 **"YES" Token**。
4. 这使系统能够绕过元数据同步延迟,直接在 CLOB 层面抓取实时买入/卖出价格。
## 4. 市场规范化结构
每个搜寻到的分段都会被规范化为以下结构,供决策引擎(Decision Engine)使用:
- `condition_id`: 用于结果判定的 UMA 条件 ID。
- `active_token_id`: 我们需要监控并买入的特定 ERC1155 Token ID。
- `group_id`: 即 `negRiskMarketID`。这让机器人知道哪些不同的温度区间是属于同一个城市的,从而进行跨区间套利或对冲分析。
- `slug`: 市场的唯一路径名,用于在 Telegram 预警中生成直接跳转链接。
## 5. 频率与缓存机制
- **搜寻频率:** 系统每 **5 分钟** 重新扫描一次新城市和新日期。
- **缓存策略:** 搜寻到的市场会存入内存缓存(`_weather_markets_cache`),以减轻 API 压力并避免触发现速限制。
---
_创建日期: 2026-02-07_
_PolyWeather 系统技术文档_
-89
View File
@@ -1,89 +0,0 @@
# 📈 PolyWeather 模拟仓 (Paper Trading) 使用指南
> ⚠️ **当前状态:功能暂停**
> 为了专注于高实时性的天气查询服务,自动模拟交易功能目前已在代码中被禁用。电报指令 `/portfolio` 暂不可用。
本系统提供全自动的模拟交易功能,让您在不投入真实资金的情况下,验证天气预测逻辑的盈利能力。
## 🛠️ 运行机制
1. **自动开仓**:
- 监控引擎在扫描中,一旦发现任何档位的 **Buy Yes****Buy No** 价格处于 **85¢ - 95¢** 区间(与城市监控报告一致),即触发买入。
- 初始本金: **$1000.00**
- 单笔投入: **动态 $3-$10**(根据四层风控策略自动调整,见下文)
- 资金检查: 余额不足时将停止开仓。
- **价格来源**: 使用真实 **Ask 价格**(实际可成交价格),而非中间价
2. **实时估值**:
- 每轮扫描结束后,系统会根据最新盘口中间价更新持仓价值。
3. **自动结项**:
- 当市场价格达到 0¢ 或 100¢(Polymarket 已结算),系统自动平仓并计算盈亏,资金回笼。
4. **数据持久化**:
- 持仓与余额保存在 `data/paper_positions.json`
## 🎯 四层风控仓位策略
系统结合 **Open-Meteo 天气预测**、**结算时间** 和 **成交量** 自动决定仓位大小:
| 条件组合 | 基础仓位 | 标签 | 说明 |
| ------------------------------- | -------- | ---------- | -------------- |
| 价格 ≥90¢ + 天气支持 + 高成交量 | **$10** | 🔥高置信 | 三重确认,重注 |
| 价格 ≥90¢ + 天气支持 | **$7** | ⭐中置信 | 双重确认 |
| 价格 ≥92¢ | **$5** | 📌价格锁定 | 纯价格锁定 |
| 其他 85-91¢ | **$3** | 💡试探 | 最小仓位试探 |
### 风控过滤规则
1. **时间衰减**:
- ≤1小时: 停止建仓 (0%)
- 1-4小时: 缩小至 40%
- 4-12小时: 缩小至 70%
- > 12小时: 100%
2. **预算上限**: 每日最高投入 $50
3. **成交量加权**: 低活跃市场额外缩减 20%
### 天气支持判断逻辑
- **买 NO**: Open-Meteo 预测温度在选项区间 **之外** (±2° 容差)
- **买 YES**: Open-Meteo 预测温度 **落入** 选项区间
### 策略建议显示
当推送包含交易信号时,会附带策略建议:
```
💡 策略建议:
• 预测温度19.0°C落在21°C区间,市场与模型一致
```
### METAR 实测数据
当天结算的市场会额外显示机场实测数据,帮助验证预测准确性:
```
✈️ 机场实测 (KORD):
🌡️ 32.0°F | 风速:12kt
🕐 观测: 14:00 UTC
```
## 📊 盈亏计算公式
- **持仓份额** = $5 / (买入价格 / 100)
- **可用余额** = 初始本金 - 累计投入总额
- **浮动盈亏** = 当前总价值 - 投入本金 ($5)
## 🤖 电报指令
您可以直接在机器人中通过以下指令查看进度:
- **/portfolio**: 实时返回当前所有“浮动”持仓的盈亏状况、历史胜率以及账户余额。
## 📁 存储文件说明
如果您需要手动清理或修改仓位,可以编辑 `data/paper_positions.json`
- `status: "OPEN"` 表示正在持仓。
- `entry_price` 以美分为单位(如 91 表示 0.91$)。
---
**蚂蚁重力 (Antigravity) 实验室**
-2
View File
@@ -54,8 +54,6 @@ py -3.11 run.py
> Local machine is for editing code and Git push only. IDE import errors are expected (dependencies not installed locally) and do not affect VPS operation.
_Note: The system is currently in **Weather Query Mode**. Legacy market monitoring and automated trading modules are suspended._
---
## 🤖 Telegram Bot Commands
-2
View File
@@ -54,8 +54,6 @@ py -3.11 run.py
> 本地笔记本**不需要安装依赖**,只用来编辑代码和 Git 推送。IDE 的 import 报错是因为本地没装依赖,不影响 VPS 运行。
_注意:系统当前处于 **天气查询模式**。主动市场监控和自动交易模块已暂停。_
---
## 🤖 Telegram 机器人指令
-4
View File
@@ -299,10 +299,6 @@ def start_bot():
parse_mode="HTML",
)
@bot.message_handler(commands=["signal", "portfolio", "status"])
def disabled_feature(message):
bot.reply_to(message, "ℹ️ 监控引擎与交易模拟功能已暂停,现仅提供天气查询服务。")
@bot.message_handler(commands=["city"])
def get_city_info(message):
"""查询指定城市的天气详情"""
-23
View File
@@ -1,23 +0,0 @@
import requests
import json
import time
def test_mgm(istno):
# 添加时间戳防止缓存
url = f"https://servis.mgm.gov.tr/web/sondurumlar?istno={istno}&_={int(time.time()*1000)}"
headers = {
"Origin": "https://www.mgm.gov.tr",
"User-Agent": "Mozilla/5.0"
}
try:
resp = requests.get(url, headers=headers)
if resp.status_code == 200:
data = resp.json()
print(json.dumps(data, indent=2, ensure_ascii=False))
else:
print(f"Error: {resp.status_code}")
except Exception as e:
print(f"Exception: {e}")
print("--- Station 17128 (Esenboğa) ---")
test_mgm(17128)
+16 -70
View File
@@ -1,68 +1,25 @@
# API Configuration
api:
polymarket:
base_url: "https://clob.polymarket.com"
ws_url: "wss://ws-subscriptions-clob.polymarket.com/ws/market"
timeout: 30
retry_attempts: 3
api_key: "019c2d40-5d23-75a6-ab33-02ae5d2a033e"
# Weather API Configuration
weather:
meteoblue_api_key: null # Set via METEOBLUE_API_KEY env var
timeout: 30
weather:
openweather:
base_url: "https://api.openweathermap.org/data/2.5"
timeout: 10
wunderground:
base_url: "https://api.weather.com/v3"
timeout: 10
visualcrossing:
base_url: "https://weather.visualcrossing.com/VisualCrossingWebServices/rest/services"
timeout: 10
# Trading Parameters
trading:
min_confidence: 0.65 # Minimum model confidence to trade
max_single_trade: 500 # Maximum single trade amount ($)
max_position_ratio: 0.25 # Maximum position as ratio of capital
max_total_exposure: 0.80 # Maximum total exposure
min_trade_size: 10 # Minimum trade size ($)
# Risk Management
risk:
max_drawdown: 0.10 # Maximum allowed drawdown (10%)
stop_loss: 0.15 # Stop loss threshold (15%)
take_profit: 0.30 # Take profit threshold (30%)
min_liquidity: 1000 # Minimum market liquidity ($)
max_slippage: 0.02 # Maximum acceptable slippage (2%)
# Analysis Parameters
analysis:
volume_threshold: 2.0 # Volume spike threshold (std dev)
large_order_threshold: 1000 # Large order detection threshold ($)
rsi_period: 14 # RSI calculation period
bollinger_period: 20 # Bollinger Bands period
bollinger_std: 2.0 # Bollinger Bands standard deviation
# Model Weights (for multi-factor decision)
weights:
statistical_prediction: 0.50
data_source_consensus: 0.15
market_volume_signal: 0.15
orderbook_analysis: 0.10
technical_indicators: 0.05
onchain_whale_signal: 0.05
# Target Markets
markets:
- id: "ankara"
city: "Ankara"
country: "Turkey"
latitude: 39.9334
longitude: 32.8597
# Target Cities
cities:
- id: "london"
city: "London"
country: "UK"
latitude: 51.5074
longitude: -0.1278
- id: "paris"
city: "Paris"
country: "France"
latitude: 48.8566
longitude: 2.3522
- id: "ankara"
city: "Ankara"
country: "Turkey"
latitude: 39.9334
longitude: 32.8597
- id: "new_york"
city: "New York"
country: "USA"
@@ -73,20 +30,9 @@ markets:
country: "USA"
latitude: 41.8781
longitude: -87.6298
- id: "paris"
city: "Paris"
country: "France"
latitude: 48.8566
longitude: 2.3522
# Logging
logging:
level: "INFO"
rotation: "10 MB"
retention: "10 days"
# Scheduler
scheduler:
data_refresh_interval: 60 # seconds
model_update_interval: 300 # seconds
risk_check_interval: 30 # seconds
-200
View File
@@ -1,200 +0,0 @@
import streamlit as st
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
from datetime import datetime, timedelta
import sys
sys.path.insert(0, '..')
st.set_page_config(
page_title="Polymarket Trading Dashboard",
page_icon="📊",
layout="wide"
)
# Custom CSS
st.markdown("""
<style>
.stMetric {
background-color: #1e1e1e;
padding: 15px;
border-radius: 10px;
}
.stMetric label {
color: #888;
}
.stMetric [data-testid="stMetricValue"] {
color: #00ff88;
}
</style>
""", unsafe_allow_html=True)
# Header
st.title("📊 Polymarket Trading Dashboard")
st.markdown("---")
# Sidebar
with st.sidebar:
st.header("⚙️ Settings")
market_id = st.text_input("Market ID", "weather-ankara-temperature")
refresh_rate = st.slider("Refresh Rate (seconds)", 10, 300, 60)
st.markdown("---")
st.header("📈 Quick Stats")
st.metric("Total PnL", "$0.00", "+0%")
st.metric("Open Positions", "0")
st.metric("Win Rate", "N/A")
# Main content
col1, col2, col3, col4 = st.columns(4)
with col1:
st.metric(
label="Current Price",
value="$0.92",
delta="+2.3%"
)
with col2:
st.metric(
label="Model Prediction",
value="7.2°C",
delta="+0.5°C"
)
with col3:
st.metric(
label="Confidence Score",
value="0.78",
delta="+0.05"
)
with col4:
st.metric(
label="Signal",
value="BUY",
delta="Strong"
)
st.markdown("---")
# Charts
col_left, col_right = st.columns(2)
with col_left:
st.subheader("📉 Price History")
# Demo price data
dates = pd.date_range(start=datetime.now() - timedelta(days=7), periods=168, freq='H')
prices = [0.85 + i * 0.0005 + (i % 24) * 0.001 for i in range(168)]
df_prices = pd.DataFrame({
'Date': dates,
'Price': prices
})
fig_price = px.line(df_prices, x='Date', y='Price',
template='plotly_dark',
color_discrete_sequence=['#00ff88'])
fig_price.update_layout(
height=300,
margin=dict(l=0, r=0, t=0, b=0)
)
st.plotly_chart(fig_price, use_container_width=True)
with col_right:
st.subheader("🌡️ Temperature Forecast")
# Demo temperature data
forecast_dates = pd.date_range(start=datetime.now(), periods=72, freq='H')
temps = [5 + (i % 24) * 0.3 + (i // 24) * 0.5 for i in range(72)]
df_temp = pd.DataFrame({
'Date': forecast_dates,
'Temperature': temps
})
fig_temp = px.line(df_temp, x='Date', y='Temperature',
template='plotly_dark',
color_discrete_sequence=['#ff6b6b'])
fig_temp.update_layout(
height=300,
margin=dict(l=0, r=0, t=0, b=0)
)
st.plotly_chart(fig_temp, use_container_width=True)
st.markdown("---")
# Decision Factors
st.subheader("🎯 Decision Factors")
factors_col1, factors_col2 = st.columns(2)
with factors_col1:
# Factor scores
factors = {
'Statistical Prediction': 0.85,
'Data Consensus': 0.90,
'Volume Signal': 0.65,
'Orderbook Analysis': 0.72,
'Technical Indicators': 0.58,
'Whale Signal': 0.45
}
fig_factors = go.Figure(go.Bar(
x=list(factors.values()),
y=list(factors.keys()),
orientation='h',
marker_color=['#00ff88' if v > 0.65 else '#ffaa00' if v > 0.4 else '#ff6b6b'
for v in factors.values()]
))
fig_factors.update_layout(
template='plotly_dark',
height=250,
margin=dict(l=0, r=0, t=0, b=0),
xaxis_title="Score",
xaxis_range=[0, 1]
)
st.plotly_chart(fig_factors, use_container_width=True)
with factors_col2:
# Order book visualization
st.markdown("**📚 Order Book**")
bids = [
{"price": 0.91, "size": 500},
{"price": 0.90, "size": 800},
{"price": 0.89, "size": 1200},
]
asks = [
{"price": 0.93, "size": 600},
{"price": 0.94, "size": 400},
{"price": 0.95, "size": 900},
]
orderbook_df = pd.DataFrame({
'Bid Price': [b['price'] for b in bids],
'Bid Size': [b['size'] for b in bids],
'Ask Price': [a['price'] for a in asks],
'Ask Size': [a['size'] for a in asks]
})
st.dataframe(orderbook_df, use_container_width=True, hide_index=True)
st.markdown("---")
# Recent Trades
st.subheader("📝 Recent Trades")
trades_df = pd.DataFrame({
'Time': ['10:30:15', '10:28:42', '10:25:11'],
'Side': ['BUY', 'BUY', 'SELL'],
'Price': ['$0.92', '$0.91', '$0.88'],
'Amount': ['$100', '$150', '$75'],
'Status': ['✅ Filled', '✅ Filled', '✅ Filled']
})
st.dataframe(trades_df, use_container_width=True, hide_index=True)
# Footer
st.markdown("---")
st.markdown("*Last updated: " + datetime.now().strftime("%Y-%m-%d %H:%M:%S") + "*")
-865
View File
@@ -1,865 +0,0 @@
import sys
import time
import os
import json
import re
from datetime import datetime, timedelta
from loguru import logger
from src.utils.config_loader import load_config
from src.utils.logger import setup_logger
from src.data_collection.polymarket_api import PolymarketClient
from src.data_collection.weather_sources import WeatherDataCollector
from src.data_collection.onchain_tracker import OnchainTracker
from src.models.statistical_model import TemperaturePredictor
from src.analysis.whale_tracker import WhaleTracker
from src.strategy.decision_engine import DecisionEngine
from src.strategy.risk_manager import RiskManager
from src.trading.paper_trader import PaperTrader
from src.utils.notifier import TelegramNotifier
def main():
# 1. 初始化配置与日志
config_data = load_config()
setup_logger(config_data.get("app", {}).get("log_level", "INFO"))
logger.info("🌟 PolyWeather 监控引擎启动中...")
# 2. 初始化核心组件
polymarket = PolymarketClient(config_data["polymarket"])
weather = WeatherDataCollector(config_data["weather"])
onchain = OnchainTracker(config_data["polymarket"], polymarket)
notifier = TelegramNotifier(config_data["telegram"])
# 3. 初始化分析与交易组件
predictor = TemperaturePredictor()
risk_manager = RiskManager(config_data.get("config", {}))
decision_engine = DecisionEngine(config_data.get("config", {}))
whale_tracker = WhaleTracker(config_data.get("config", {}), onchain)
paper_trader = PaperTrader()
# 发送启动通知
notifier._send_message(
"🚀 <b>Polymarket 天气监控系统启动成功</b>\n正在扫描 12 个核心城市的最高温市场..."
)
# 信号记忆(持久化到文件)
pushed_signals = {}
SIGNALS_FILE = "data/pushed_signals.json"
if os.path.exists(SIGNALS_FILE):
try:
with open(SIGNALS_FILE, "r", encoding="utf-8") as f:
pushed_signals = json.load(f)
logger.info(f"已加载历史推送记录,共 {len(pushed_signals)}")
except:
pushed_signals = {}
# 确保data目录存在
if not os.path.exists("data"):
os.makedirs("data")
location_cache = {}
# 价格历史追踪(用于计算趋势)
PRICE_HISTORY_FILE = "data/price_history.json"
price_history = {}
if os.path.exists(PRICE_HISTORY_FILE):
try:
with open(PRICE_HISTORY_FILE, "r", encoding="utf-8") as f:
price_history = json.load(f)
except:
price_history = {}
try:
while True:
logger.info("--- 开启新一轮全量动态监控 (自动搜寻所有天气市场) ---")
cached_signals = {}
all_markets_cache = {}
# 1. 直接从 Polymarket 获取所有天气合约
all_weather_markets = polymarket.get_weather_markets()
# 1.5 尝试通过slug获取可能遗漏的市场(如部分结算的市场)
special_slugs = []
for slug in special_slugs:
event = polymarket.get_event_by_slug(slug)
if event:
title = event.get("title", "")
logger.info(f"通过slug找到特殊事件: {title}")
# 提取城市名
city = weather.extract_city_from_question(title)
if not city:
city = "Unknown"
# 将该事件的所有市场添加到列表
for m in event.get("markets", []):
# 检查是否已存在
c_id = m.get("conditionId")
if not any(
existing.get("condition_id") == c_id
for existing in all_weather_markets
):
all_weather_markets.append(
{
"condition_id": c_id,
"question": m.get("groupItemTitle")
or m.get("question"),
"active_token_id": m.get("activeTokenId"),
"tokens": m.get("clobTokenIds"),
"prices": m.get("outcomePrices"),
"event_title": title,
"slug": slug,
"city": city, # 提前标记城市
}
)
logger.debug(f"添加特殊市场: {m.get('groupItemTitle')}")
if not all_weather_markets:
logger.warning("当前 Polymarket 似乎没有任何活跃的天气市场,等待中...")
time.sleep(300)
continue
# 2. 批量同步盘口价格 (优化:为每个档位获取其对应的真实 Token 价格)
token_price_map = {}
price_requests = []
for m in all_weather_markets:
ts = m.get("tokens", [])
if isinstance(ts, str):
try:
ts = json.loads(ts)
except:
ts = []
active_tid = m.get("active_token_id")
# 智能识别买入/买否 Token
if active_tid and isinstance(ts, list):
# 获取该档位的买入价 (Ask)
price_requests.append({"token_id": active_tid, "side": "ask"})
if len(ts) == 2:
# 传统的二选一,直接获取 No Token 的 Ask
no_tid = ts[1] if ts[0] == active_tid else ts[0]
price_requests.append({"token_id": no_tid, "side": "ask"})
else:
# 多选一,需要用 1 - Bid(Yes) 来模拟 Buy No
price_requests.append({"token_id": active_tid, "side": "bid"})
if price_requests:
logger.info(f"正在同步 {len(price_requests)} 个档位的真实盘口价格...")
token_price_map = polymarket.get_multiple_prices(price_requests)
logger.info(f"价格同步完成,成功获取 {len(token_price_map)} 个实时报价")
# 3. 按城市分组(按condition_id去重)
markets_by_city = {}
seen_condition_ids = set() # Initialize seen_condition_ids here
for i, m in enumerate(all_weather_markets):
# Use condition_id + active_token_id as unique key to support multi-bracket markets
unique_market_key = f"{m.get('condition_id')}_{m.get('active_token_id')}"
if unique_market_key in seen_condition_ids:
continue
seen_condition_ids.add(unique_market_key)
# 注入实时批量价格
ts = m.get("tokens", [])
if isinstance(ts, str):
try:
ts = json.loads(ts)
except:
ts = []
active_tid = m.get("active_token_id")
if active_tid and isinstance(ts, list):
m["buy_yes_live"] = token_price_map.get(f"{active_tid}:ask")
if len(ts) == 2:
no_tid = ts[1] if ts[0] == active_tid else ts[0]
m["buy_no_live"] = token_price_map.get(f"{no_tid}:ask")
else:
# 1 - Bid(Yes) = Ask(No)
bid_val = token_price_map.get(f"{active_tid}:bid")
if bid_val:
m["buy_no_live"] = 1.0 - bid_val
# 优先使用发现阶段已经识别出的城市名
city = m.get("city")
# 如果发现阶段没识别出,再尝试从问题文本或 Slug 提取
if not city or city == "Unknown":
full_context = f"{m.get('event_title', '')} {m.get('question', '')} {m.get('slug', '')}"
city = weather.extract_city_from_question(full_context)
if i < 5:
logger.debug(
f"分析合约 {i}: City='{city}' | Title='{m.get('event_title')}"
)
if not city:
continue
if city not in markets_by_city:
markets_by_city[city] = []
markets_by_city[city].append(m)
logger.info(
f"动态发现 {len(markets_by_city)} 个受监控城市,共 {len(all_weather_markets)} 个合约"
)
# 3. 逐个城市分析
for city, city_markets in markets_by_city.items():
try:
# 获取/缓存坐标
if city not in location_cache:
coords = weather.get_coordinates(city)
if not coords:
continue
location_cache[city] = coords
logger.info(
f"📍 城市定位成功: {city} -> ({coords['lat']}, {coords['lon']})"
)
loc = location_cache[city]
# A. 获取实时天气共识
weather_data = weather.fetch_all_sources(
city, lat=loc["lat"], lon=loc["lon"]
)
consensus = weather.check_consensus(weather_data)
if not consensus.get("consensus"):
continue
temp_unit = weather_data.get("open-meteo", {}).get(
"unit", "celsius"
)
temp_symbol = "°F" if temp_unit == "fahrenheit" else "°C"
logger.info(
f"☁️ {city} 当前气温: {consensus['average_temp']}{temp_symbol} (unit={temp_unit}) | 监控合约: {len(city_markets)}"
)
# --- 本城市汇总预警缓存 ---
city_alerts = []
city_local_time = None
city_total_vol = 0
city_pred_high = None
city_target_date = None
city_strategy_tips = []
# B. 遍历该城市所有合约
for market in city_markets:
market_id = market.get("condition_id")
question = market.get("question", "未知市场")
event_title = market.get("event_title", "")
# 累计城市总成交量
vol_raw = market.get("volume", 0)
if isinstance(vol_raw, str):
try:
vol_raw = float(
vol_raw.replace("$", "").replace(",", "")
)
except:
vol_raw = 0
city_total_vol += vol_raw
# 识别该合约的目标日期
target_date = weather.extract_date_from_title(
event_title
) or weather.extract_date_from_title(question)
ref_temp = consensus["average_temp"]
if target_date:
daily_data = weather_data.get("open-meteo", {}).get(
"daily", {}
)
if daily_data:
dates = daily_data.get("time", [])
max_temps = daily_data.get("temperature_2m_max", [])
for idx, d_str in enumerate(dates):
if target_date == d_str:
ref_temp = max_temps[idx]
break
# --- 价格获取逻辑 (增强版) ---
# 使用 token_price_map 获取实时数据
active_tid = market.get("active_token_id")
ts = market.get("tokens", [])
if isinstance(ts, str):
ts = json.loads(ts)
buy_yes_price = None
buy_no_price = None
bid_yes_price = None
if len(ts) == 2:
# 传统二选一市场 (Yes/No Token 独立)
buy_yes_price = token_price_map.get(f"{ts[0]}:ask")
buy_no_price = token_price_map.get(f"{ts[1]}:ask")
bid_yes_price = token_price_map.get(f"{ts[0]}:bid")
elif active_tid:
# 多选一市场 (单 Token 对应一个档位)
buy_yes_price = token_price_map.get(f"{active_tid}:ask")
bid_yes_price = token_price_map.get(f"{active_tid}:bid")
if bid_yes_price is not None:
buy_no_price = 1.0 - bid_yes_price
# 兜底概率计算
current_prob = (
(buy_yes_price + bid_yes_price) / 2
if (buy_yes_price and bid_yes_price)
else (buy_yes_price or 0.5)
)
if buy_no_price is None:
buy_no_price = 1.0 - current_prob
# 计算价格趋势
prev_data = price_history.get(market_id, {})
prev_prob = prev_data.get("price", current_prob)
prob_change = (current_prob - prev_prob) * 100
trend_str = (
f"{abs(prob_change):.0f}%"
if prob_change > 0.5
else (
f"{abs(prob_change):.0f}%"
if prob_change < -0.5
else ""
)
)
# 更新历史缓存
price_history[market_id] = {
"price": current_prob,
"timestamp": datetime.now().isoformat(),
}
# --- 预警收集 (自动推送逻辑) ---
# 严格触发条件: 价格必须处于 85-95¢ 区间 (真正的高概率信号)
yes_in_range = buy_yes_price and 0.85 <= buy_yes_price <= 0.95
no_in_range = buy_no_price and 0.85 <= buy_no_price <= 0.95
# 50¢ 保护:价格接近 50% 说明市场无明确方向,跳过
is_undecided = 0.45 <= current_prob <= 0.55
if (yes_in_range or no_in_range) and not is_undecided:
alert_key = f"alert_{market_id}_{int(current_prob * 100)}"
if alert_key not in pushed_signals:
# 获取温度符号(在此处定义以便后续使用)
temp_unit = weather_data.get("open-meteo", {}).get(
"unit", "celsius"
)
temp_symbol = (
"°F" if temp_unit == "fahrenheit" else "°C"
)
# 预测偏差分析
if ref_temp:
city_pred_high = ref_temp # 记录到城市概览
temp_match = re.search(
r"(\d+)(?:-(\d+))?°[FC]", question
)
if temp_match:
low_b = int(temp_match.group(1))
high_b = (
int(temp_match.group(2))
if temp_match.group(2)
else low_b
)
diff = ref_temp - ((low_b + high_b) / 2)
# 偏差信息将在后面构建 msg 时统一添加
# 生成策略建议:仅保留模型一致提示
if abs(diff) < 2 and current_prob > 0.7:
city_strategy_tips.append(
f"预测温度{ref_temp}{temp_symbol}落在{question}区间,市场与模型一致"
)
# 模拟下单 - 使用 Ask 价格(实际可成交价格)
if buy_yes_price and buy_yes_price > 0.5:
trigger_side = "Buy Yes"
trigger_price = int(buy_yes_price * 100)
else:
trigger_side = "Buy No"
trigger_price = (
int(buy_no_price * 100)
if buy_no_price
else int((1 - current_prob) * 100)
)
# 构建预测文本
forecast_text = (
f"{ref_temp}{temp_symbol}" if ref_temp else "N/A"
)
# 构建简约版消息
side_display = (
"Buy No" if trigger_side == "Buy No" else "Buy Yes"
)
msg = f"{question} ({target_date}): {side_display} {trigger_price}¢ | 预测:{forecast_text}"
success = paper_trader.open_position(
market_id=market_id,
city=city,
option=question,
price=trigger_price,
side="YES" if trigger_side == "Buy Yes" else "NO",
amount_usd=5.0,
target_date=target_date,
predicted_temp=ref_temp,
)
# 添加模拟交易标签
if success:
msg += " [🛒 $5.0 💡试探]"
city_alerts.append(
{
"market": target_date or "今日",
"msg": msg,
"bought": success,
"amount": 5.0,
"confidence": "💡试探",
}
)
pushed_signals[alert_key] = time.time()
if target_date:
city_target_date = target_date
# C. 准备缓存数据
temp_unit = weather_data.get("open-meteo", {}).get(
"unit", "celsius"
)
temp_symbol = "°F" if temp_unit == "fahrenheit" else "°C"
city_local_time = (
weather_data.get("open-meteo", {})
.get("current", {})
.get("local_time")
)
current_price = buy_yes_price if buy_yes_price else 0.5
# 计算价格趋势
prev_data = price_history.get(market_id, {})
prev_price = prev_data.get("price", current_price)
price_change_pct = (
((current_price - prev_price) / prev_price * 100)
if prev_price > 0
else 0
)
# 更新价格历史缓存
price_history[market_id] = {
"price": current_price,
"timestamp": datetime.now().isoformat(),
}
cache_entry = {
"city": city,
"full_title": event_title,
"option": question,
"prediction": f"{ref_temp}{temp_symbol}",
"price": int(current_price * 100),
"buy_yes": int(buy_yes_price * 100) if buy_yes_price else 0,
"buy_no": int(buy_no_price * 100) if buy_no_price else 0,
"url": f"https://polymarket.com/event/{market.get('slug')}",
"local_time": city_local_time,
"target_date": target_date,
"score": 0,
"rationale": "ACTIVE",
"trend": round(price_change_pct, 1),
}
# --- 最终过滤器 (拦截垃圾信号) ---
# 1. 过滤已锁定价格 (>= 98.5c)
if (buy_yes_price and buy_yes_price >= 0.985) or (
buy_no_price and buy_no_price >= 0.985
):
cache_entry["rationale"] = "ENDED"
all_markets_cache[market_id] = cache_entry
continue
# 2. 过滤已过期日期 (动态获取当前日期)
current_today = datetime.now().strftime("%Y-%m-%d")
if target_date and target_date < current_today:
cache_entry["rationale"] = "EXPIRED"
all_markets_cache[market_id] = cache_entry
continue
# 3. 评分计算
try:
signal = decision_engine.calculate_signal(
model_prediction=predictor.predict_ensemble([ref_temp]),
market_data={
"orderbook": {},
"price_history": [current_price],
"transactions": [],
},
weather_consensus={"average_temp": ref_temp},
whale_activity=None,
)
cache_entry["score"] = signal.get("final_score", 0)
cache_entry["rationale"] = signal.get(
"recommendation", "ACTIVE"
)
except Exception as e:
logger.error(f"计算信号失败 [{market_id}]: {e}")
cache_entry["score"] = 0
cache_entry["rationale"] = "ERROR"
all_markets_cache[market_id] = cache_entry
# --- 预警收集 (自动推送逻辑) ---
if (buy_yes_price and 0.85 <= buy_yes_price <= 0.95) or (
buy_no_price and 0.85 <= buy_no_price <= 0.95
):
alert_key = f"alert_{market_id}_range_85_95"
if alert_key not in pushed_signals:
# --- 基础参数识别 ---
is_categorical = len(ts) > 2 and active_tid
if is_categorical:
# 语义转换逻辑保持一致
if buy_no_price and buy_no_price >= 0.85:
trigger_side = "Buy No" # 直接统一为 Buy No
trigger_price = int(buy_no_price * 100)
else:
trigger_side = "Buy Yes"
trigger_price = int(buy_yes_price * 100)
else:
trigger_side = (
"Buy Yes" if buy_yes_price >= 0.85 else "Buy No"
)
trigger_price = (
int(buy_yes_price * 100)
if trigger_side == "Buy Yes"
else int(buy_no_price * 100)
)
# --- 智能动态仓位计算 ---
# 1. 获取 Open-Meteo 对目标日期的最高温预测
predicted_high = None
weather_supports = False
daily_data = weather_data.get("open-meteo", {}).get(
"daily", {}
)
if daily_data and target_date:
dates = daily_data.get("time", [])
max_temps = daily_data.get("temperature_2m_max", [])
for idx, d_str in enumerate(dates):
if target_date == d_str and idx < len(
max_temps
):
predicted_high = max_temps[idx]
break
# 2. 判断天气预测是否支持当前方向
if predicted_high is not None:
# 解析选项的温度范围 (例如 "40-41°F" 或 "32°F or below")
temp_match = re.search(
r"(\d+)(?:-(\d+))?°[FC]", question
)
if temp_match:
low_bound = int(temp_match.group(1))
high_bound = (
int(temp_match.group(2))
if temp_match.group(2)
else low_bound
)
# 如果买 NO,天气预测应该在这个区间之外
if trigger_side == "Buy No":
weather_supports = (
predicted_high < low_bound - 2
) or (predicted_high > high_bound + 2)
else: # 买 YES
weather_supports = (
low_bound - 2
<= predicted_high
<= high_bound + 2
)
# 3. 获取成交量信息
market_volume = market.get("volume", 0)
if isinstance(market_volume, str):
try:
market_volume = float(
market_volume.replace("$", "").replace(
",", ""
)
)
except:
market_volume = 0
high_volume = market_volume >= 5000 # $5000+ 算高成交量
# --- Pro 级仓位决策系统 ---
# 1. 计算离结算剩余小时数 (假设气温市场在目标日期晚上 23:59 结算)
hours_to_settle = 24.0
if target_date:
try:
settle_dt = datetime.strptime(
f"{target_date} 23:59:59",
"%Y-%m-%d %H:%M:%S",
)
now_utc = datetime.utcnow()
diff = settle_dt - now_utc
hours_to_settle = diff.total_seconds() / 3600.0
except:
pass
# 2. 计算相对成交量比例
total_daily_vol = sum(
[
float(
str(m.get("volume", 0))
.replace("$", "")
.replace(",", "")
)
for m in city_markets
if (
weather.extract_date_from_title(
m.get("event_title", "")
)
or weather.extract_date_from_title(
m.get("question", "")
)
)
== target_date
]
)
market_vol = float(
str(market.get("volume", 0))
.replace("$", "")
.replace(",", "")
)
is_rel_high_vol = (
(market_vol / total_daily_vol > 0.3)
if total_daily_vol > 0
else False
)
# 3. 基础意向仓位 (基于置信度)
base_pos = 3.0 # 默认探路
confidence_tag = "💡试探"
if (
trigger_price >= 90
and weather_supports
and high_volume
):
base_pos, confidence_tag = 10.0, "🔥高置信"
elif trigger_price >= 90 and weather_supports:
base_pos, confidence_tag = 7.0, "⭐中置信"
elif trigger_price >= 92:
base_pos, confidence_tag = 5.0, "📌价格锁定"
# 4. 仓位决策
amount_usd, risk_reason = (
risk_manager.calculate_position_size(
base_confidence_usd=base_pos,
hours_to_settle=hours_to_settle,
is_high_relative_volume=is_rel_high_vol,
)
)
logger.info(
f"【Pro仓位】{city} {question} | "
f"基础:{base_pos}$ -> 最终:{amount_usd}$ | 原因:{risk_reason} | "
f"剩:{hours_to_settle:.1f}h"
)
# --- 模拟交易触发逻辑 ---
if amount_usd > 0:
side = "YES" if trigger_side == "Buy Yes" else "NO"
success = paper_trader.open_position(
market_id=market_id,
city=city,
option=question,
price=trigger_price,
side=side,
amount_usd=amount_usd,
target_date=target_date,
predicted_temp=predicted_high,
)
if success:
risk_manager.record_trade(amount_usd)
else:
# 如果被风控拦截(金额为0),则不进行任何推送,避免刷屏
success = False
logger.info(
f"Skipping alert for {question}: {risk_reason}"
)
continue
# 构建预测温度显示文本
temp_unit = weather_data.get("open-meteo", {}).get(
"unit", "celsius"
)
temp_symbol = (
"°F" if temp_unit == "fahrenheit" else "°C"
)
forecast_text = (
f"{predicted_high}{temp_symbol}"
if predicted_high
else "N/A"
)
# 构建简约版消息: ⚡ {question} ({date}): {side} {price}¢ | 预测:{forecast} [🛒 ${amount} {tag}]
side_display = trigger_side
msg = (
f"{question} ({target_date}): {side_display} {trigger_price}¢ | "
f"预测:{forecast_text} [🛒 ${amount_usd} {confidence_tag}]"
)
city_alerts.append(
{
"type": "price",
"market": f"{target_date or '今日'}",
"msg": msg,
"bought": success,
"amount": amount_usd,
"confidence": confidence_tag,
}
)
pushed_signals[alert_key] = time.time()
# 3. 信号暂存
cached_signals[market_id] = cache_entry
# E. 统一发送城市汇总通知 (使用新 Pro 模板)
if city_alerts:
# 去重策略建议
unique_tips = list(dict.fromkeys(city_strategy_tips))
# 获取 METAR 数据(仅当天结算的市场才显示)
today_str = datetime.now().strftime("%Y-%m-%d")
# 检查是否有当天结算的市场
has_today_market = any(
a.get("market") == today_str or a.get("market") == "今日"
for a in city_alerts
)
metar_data = (
weather_data.get("metar") if has_today_market else None
)
# notifier.send_combined_alert(
# city=city,
# alerts=city_alerts,
# local_time=city_local_time,
# forecast_temp=f"{city_pred_high}{temp_symbol}"
# if city_pred_high
# else "N/A",
# total_volume=city_total_vol,
# brackets_count=len(city_markets),
# strategy_tips=unique_tips,
# metar_data=metar_data,
# )
except Exception as e:
logger.error(f"分析城市 {city} 时出错: {e}")
# --- 每处理完一个城市,立即更新 JSON 文件 ---
try:
# --- 周期性结算:保存高价值信号 ---
active_signals = []
for mid, entry in all_markets_cache.items():
# Relaxed filtering: Let the bot decide, but mark ENDED
rationale = entry.get("rationale")
if rationale == "ERROR":
continue
target_dt = entry.get("target_date")
# Only filter out truly ancient history
if target_dt and target_dt < "2026-02-01":
continue
active_signals.append(entry)
# 按分数排序
active_signals.sort(key=lambda x: x.get("score", 0), reverse=True)
with open("data/active_signals.json", "w", encoding="utf-8") as f:
json.dump(active_signals, f, ensure_ascii=False, indent=4)
logger.info(
f"已更新活跃信号库,包含 {len(active_signals)} 个有效信号。"
)
# 2. 更新全量市场缓存
try:
with open("data/all_markets.json", "r", encoding="utf-8") as f:
existing_markets = json.load(f)
except:
existing_markets = {}
existing_markets.update(all_markets_cache)
# 清理过期日期
today_str = datetime.now().strftime("%Y-%m-%d")
cleaned_markets = {}
for k, v in existing_markets.items():
t_date = v.get("target_date")
if not t_date or t_date >= today_str:
cleaned_markets[k] = v
with open("data/all_markets.json", "w", encoding="utf-8") as f:
json.dump(cleaned_markets, f, ensure_ascii=False, indent=2)
# 3. 保存推送记录
with open("data/pushed_signals.json", "w", encoding="utf-8") as f:
json.dump(pushed_signals, f, ensure_ascii=False)
# 3.5 保存价格历史(用于趋势计算)
with open(PRICE_HISTORY_FILE, "w", encoding="utf-8") as f:
json.dump(price_history, f, ensure_ascii=False)
# --- 4. 更新模拟仓位盈亏 ---
price_snapshot = {}
for mid, entry in all_markets_cache.items():
price_snapshot[mid] = {"price": entry["price"]}
paper_trader.update_pnl(price_snapshot)
# --- 5. 每日收益总结推送 (北京时间 23:55 - 00:05 之间发送) ---
now_bj = datetime.utcnow() + timedelta(hours=8)
if now_bj.hour == 23 and now_bj.minute >= 50:
summary_key = f"daily_pnl_{now_bj.strftime('%Y%m%d')}"
if summary_key not in pushed_signals:
# 构造总结消息
total_cost = 0
total_pnl = 0
data = paper_trader._load_data()
pos_list = data.get("positions", {})
if pos_list:
report = [
f"📊 <b>每日模拟仓结算总结 ({now_bj.strftime('%Y-%m-%d')})</b>\n"
+ "" * 15
]
for p in pos_list.values():
if p["status"] == "OPEN":
total_cost += p["cost_usd"]
total_pnl += p.get("pnl_usd", 0)
report.append(
f"💳 可用余额: <b>${data.get('balance', 0):.2f}</b>"
)
report.append(
f"💰 今日累计投入: <b>${total_cost:.2f}</b>"
)
report.append(
f"📈 累计浮动盈亏: <b>{total_pnl:+.2f}$</b>"
)
# notifier._send_message("\n".join(report))
pushed_signals[summary_key] = time.time()
except Exception as e:
logger.error(f"即时保存数据失败: {e}")
logger.info("本轮扫描结束。等待 5 分钟...")
time.sleep(300)
except KeyboardInterrupt:
logger.info("收到关机指令,正在退出...")
except Exception as e:
logger.exception(f"系统运行出错: {e}")
if __name__ == "__main__":
main()
+9 -33
View File
@@ -1,48 +1,24 @@
import threading
import time
import sys
import subprocess
import os
import sys
from loguru import logger
def run_monitor():
"""启动监控引擎模块 (main.py)"""
logger.info("📡 正在启动后台监控引擎 (主动预警模式)...")
cmd = [sys.executable, "main.py"]
subprocess.run(cmd)
def run_bot():
"""启动电报交互模块 (bot_listener.py)"""
logger.info("🤖 正在启动电报指令监听器 (被动查询模式)...")
cmd = [sys.executable, "bot_listener.py"]
# 设置工作目录,确保导入正常
subprocess.run(cmd, cwd=os.getcwd())
def main():
logger.info("🌟 PolyWeather 全功能系统正在初始化...")
# 创建共享文件夹 (如果不存在)
if not os.path.exists("data"):
os.makedirs("data")
logger.info("🌡️ PolyWeather 天气查询机器人启动中...")
# 创建两个线程并行运行
monitor_thread = threading.Thread(target=run_monitor, daemon=True)
bot_thread = threading.Thread(target=run_bot, daemon=True)
# 创建数据目录
os.makedirs("data", exist_ok=True)
# 启动线程
# monitor_thread.start()
bot_thread.start()
logger.success("🚀 系统已上线(天气查询模式)!")
logger.info("已暂停监控引擎和自动发现市场功能。")
logger.info("现在仅支持直接查询各城市实时天气与 Open-Meteo 预测。")
# 直接运行 bot_listener
cmd = [sys.executable, "bot_listener.py"]
logger.success("🚀 已上线!等待 Telegram 指令...")
try:
# 保持主进程运行
while True:
time.sleep(1)
subprocess.run(cmd, cwd=os.getcwd())
except KeyboardInterrupt:
logger.warning("停止运行...")
if __name__ == "__main__":
main()
View File
-95
View File
@@ -1,95 +0,0 @@
from loguru import logger
class OrderbookAnalyzer:
"""
分析目标: 评估市场供需平衡和流动性
"""
def __init__(self, config=None):
self.config = config or {}
self.wall_threshold = self.config.get("wall_threshold", 500) # 单笔订单超过此值为墙
logger.info("Initializing Orderbook Analyzer...")
def assess_liquidity(self, orderbook, side="ask"):
"""
分析流动性深度 (基于前 3 档)
"""
orders = orderbook.get('asks' if side == "ask" else 'bids', [])
if not orders:
return "枯竭", 0
# 前 3 档总量 (Polymarket 通常返回价格字符串)
depth = sum(float(o.get("size", 0)) for o in orders[:3])
if depth < 50:
return "稀薄", depth
elif depth < 500:
return "正常", depth
else:
return "充裕", depth
def analyze(self, orderbook):
"""
增强版订单簿分析:集成深度与 Spread 评估
"""
bids = orderbook.get('bids', [])
asks = orderbook.get('asks', [])
if not bids or not asks:
return {
"signal": "NEUTRAL",
"confidence": 0.0,
"tradeable": False,
"reason": "缺乏双边报价",
"liquidity": "枯竭",
"spread": 1.0
}
# 1. 计算核心指标
best_bid = float(bids[0].get('price', 0))
best_ask = float(asks[0].get('price', 0))
spread = abs(best_ask - best_bid)
mid_price = (best_ask + best_bid) / 2
# 2. 评估流动性
ask_liq, ask_depth = self.assess_liquidity(orderbook, "ask")
bid_liq, bid_depth = self.assess_liquidity(orderbook, "bid")
# 3. 交易可行性判定 (Spread <= 10c 且 深度 >= $50)
is_tradeable = (spread <= 0.10) and (ask_depth >= 50 or bid_depth >= 50)
# 4. Imbalance 计算
bid_volume = sum([float(b.get('size', 0)) for b in bids])
ask_volume = sum([float(a.get('size', 0)) for a in asks])
imbalance = bid_volume / ask_volume if ask_volume > 0 else 0
result = {
"best_bid": best_bid,
"best_ask": best_ask,
"mid_price": mid_price,
"spread": round(spread, 4),
"ask_depth": round(ask_depth, 2),
"bid_depth": round(bid_depth, 2),
"liquidity": ask_liq if ask_depth < bid_depth else bid_liq,
"tradeable": is_tradeable,
"imbalance": imbalance,
"signal": "NEUTRAL",
"confidence": 0.5
}
# 5. 信号修正
if is_tradeable:
if imbalance > 2.5:
result["signal"] = "BULLISH"
result["confidence"] = 0.75
elif imbalance < 0.4:
result["signal"] = "BEARISH"
result["confidence"] = 0.75
else:
result["confidence"] = 0.1 # 不建议交易
return result
def analyze_orderbook(orderbook):
"""兼容旧接口的便捷函数"""
analyzer = OrderbookAnalyzer()
return analyzer.analyze(orderbook)
-146
View File
@@ -1,146 +0,0 @@
import numpy as np
from loguru import logger
class TechnicalIndicators:
"""
技术指标计算 - RSI, 布林带等
"""
def __init__(self):
logger.info("Initializing Technical Indicators...")
def calculate_rsi(self, prices: list, period: int = 14) -> float:
"""
计算相对强弱指标 (RSI)
Args:
prices: 价格历史列表
period: RSI周期,默认14
Returns:
float: RSI值 (0-100)
"""
if len(prices) < period + 1:
logger.debug("Insufficient data for RSI calculation")
return 50.0 # 返回中性值
prices = np.array(prices)
deltas = np.diff(prices)
gains = np.where(deltas > 0, deltas, 0)
losses = np.where(deltas < 0, -deltas, 0)
avg_gain = np.mean(gains[-period:])
avg_loss = np.mean(losses[-period:])
if avg_loss == 0:
return 100.0
rs = avg_gain / avg_loss
rsi = 100 - (100 / (1 + rs))
logger.debug(f"RSI({period}): {rsi:.2f}")
return rsi
def calculate_bollinger_bands(self, prices: list, period: int = 20, std_dev: float = 2.0) -> dict:
"""
计算布林带
Args:
prices: 价格历史列表
period: 移动平均周期
std_dev: 标准差倍数
Returns:
dict: 包含上轨、中轨、下轨
"""
if len(prices) < period:
logger.debug("Insufficient data for Bollinger Bands")
return {"upper": None, "middle": None, "lower": None}
prices = np.array(prices[-period:])
middle = np.mean(prices)
std = np.std(prices)
upper = middle + std_dev * std
lower = middle - std_dev * std
return {
"upper": upper,
"middle": middle,
"lower": lower,
"std": std
}
def calculate_momentum(self, prices: list, period: int = 10) -> float:
"""
计算价格动量
Args:
prices: 价格历史
period: 动量周期
Returns:
float: 动量值 (当前价格 / N周期前价格 - 1)
"""
if len(prices) < period + 1:
return 0.0
current = prices[-1]
past = prices[-period - 1]
if past == 0:
return 0.0
momentum = (current / past) - 1
return momentum
def get_signal(self, prices: list) -> dict:
"""
综合技术指标信号
Returns:
dict: 包含信号和分数
"""
rsi = self.calculate_rsi(prices)
bb = self.calculate_bollinger_bands(prices)
momentum = self.calculate_momentum(prices)
# RSI信号
if rsi > 70:
rsi_signal = "OVERBOUGHT"
rsi_score = 0.3 # 超买,看跌
elif rsi < 30:
rsi_signal = "OVERSOLD"
rsi_score = 0.8 # 超卖,看涨
else:
rsi_signal = "NEUTRAL"
rsi_score = 0.5
# 布林带信号
if bb["upper"] and len(prices) > 0:
current_price = prices[-1]
if current_price > bb["upper"]:
bb_signal = "ABOVE_UPPER"
bb_score = 0.7 # 突破上轨,强势
elif current_price < bb["lower"]:
bb_signal = "BELOW_LOWER"
bb_score = 0.3 # 跌破下轨,弱势
else:
bb_signal = "WITHIN_BANDS"
bb_score = 0.5
else:
bb_signal = "NO_DATA"
bb_score = 0.5
# 综合分数
combined_score = (rsi_score * 0.5 + bb_score * 0.3 +
(0.5 + momentum * 2) * 0.2) # momentum 转换为 0-1
combined_score = max(0, min(1, combined_score))
return {
"rsi": {"value": rsi, "signal": rsi_signal, "score": rsi_score},
"bollinger": {"bands": bb, "signal": bb_signal, "score": bb_score},
"momentum": momentum,
"combined_score": combined_score
}
-135
View File
@@ -1,135 +0,0 @@
import numpy as np
from loguru import logger
class VolumeAnalyzer:
"""
交易量异常检测 - 识别聪明钱和市场转折点
"""
def __init__(self, config=None):
self.config = config or {}
self.volume_threshold = self.config.get("volume_threshold", 2.0) # 2倍标准差
self.large_order_threshold = self.config.get("large_order_threshold", 1000) # $1000
logger.info("Initializing Volume Analyzer...")
def detect_volume_spike(self, volume_history: list) -> dict:
"""
检测成交量异常放大
Args:
volume_history: 历史成交量列表
Returns:
dict: 包含信号和置信度
"""
if len(volume_history) < 24:
return {"signal": "INSUFFICIENT_DATA", "score": 0.5}
recent_volume = np.array(volume_history[-24:]) # 最近24小时
historical_volume = np.array(volume_history[:-24])
if len(historical_volume) == 0:
return {"signal": "INSUFFICIENT_DATA", "score": 0.5}
avg_volume = np.mean(historical_volume)
std_volume = np.std(historical_volume)
recent_avg = np.mean(recent_volume)
# 计算Z-score
if std_volume > 0:
z_score = (recent_avg - avg_volume) / std_volume
else:
z_score = 0
logger.debug(f"Volume Z-score: {z_score:.2f}")
if z_score > self.volume_threshold:
return {
"signal": "VOLUME_SPIKE",
"score": min(0.9, 0.5 + z_score * 0.1),
"z_score": z_score,
"interpretation": "成交量异常放大,可能有新信息进入市场"
}
elif z_score < -self.volume_threshold:
return {
"signal": "VOLUME_DRY",
"score": 0.3,
"z_score": z_score,
"interpretation": "成交量萎缩,市场观望"
}
return {"signal": "NORMAL", "score": 0.5, "z_score": z_score}
def detect_large_orders(self, transactions: list) -> dict:
"""
检测大额订单 (聪明钱信号)
Args:
transactions: 交易列表,每个包含 size, side, price
Returns:
dict: 大额订单分析结果
"""
large_buys = []
large_sells = []
for tx in transactions:
size = tx.get("size", 0)
side = tx.get("side", "").upper()
if size >= self.large_order_threshold:
if side == "BUY":
large_buys.append(tx)
elif side == "SELL":
large_sells.append(tx)
total_large_buy = sum(t.get("size", 0) for t in large_buys)
total_large_sell = sum(t.get("size", 0) for t in large_sells)
logger.debug(f"Large buys: ${total_large_buy:.2f}, Large sells: ${total_large_sell:.2f}")
if total_large_buy > total_large_sell * 2:
return {
"signal": "SMART_MONEY_BUY",
"score": 0.8,
"large_buy_volume": total_large_buy,
"large_sell_volume": total_large_sell,
"interpretation": "大户在积极买入,跟随机会"
}
elif total_large_sell > total_large_buy * 2:
return {
"signal": "SMART_MONEY_SELL",
"score": 0.2,
"large_buy_volume": total_large_buy,
"large_sell_volume": total_large_sell,
"interpretation": "大户在抛售,风险警告"
}
return {
"signal": "NEUTRAL",
"score": 0.5,
"large_buy_volume": total_large_buy,
"large_sell_volume": total_large_sell
}
def analyze(self, volume_history: list, transactions: list = None) -> dict:
"""
综合分析交易量
"""
volume_signal = self.detect_volume_spike(volume_history)
if transactions:
order_signal = self.detect_large_orders(transactions)
else:
order_signal = {"signal": "NO_DATA", "score": 0.5}
# 综合评分
combined_score = (volume_signal.get("score", 0.5) * 0.6 +
order_signal.get("score", 0.5) * 0.4)
return {
"volume_signal": volume_signal,
"order_signal": order_signal,
"combined_score": combined_score
}
-59
View File
@@ -1,59 +0,0 @@
from loguru import logger
from typing import List, Dict
from src.data_collection.onchain_tracker import OnchainTracker
class WhaleTracker:
"""
大户行为分析模块
"""
def __init__(self, config: dict, tracker: OnchainTracker):
self.config = config
self.tracker = tracker
logger.info("Initializing Whale Tracker...")
def analyze_market_whales(self, market_id: str) -> Dict:
"""
分析特定市场的鲸鱼行为
"""
large_trades = self.tracker.get_large_transactions(market_id)
if not large_trades:
return {"bullish": False, "signal": "NEUTRAL", "reason": "No whale activity detected"}
buy_value = 0
sell_value = 0
for trade in large_trades:
side = trade.get("side", "").upper()
value = trade.get("value", 0)
if side == "BUY":
buy_value += value
else:
sell_value += value
# 判断情绪
if buy_value > sell_value * 2:
return {
"bullish": True,
"signal": "STRONG_ACCUMULATION",
"buy_value": buy_value,
"sell_value": sell_value,
"reason": "Whales are heavily buying"
}
elif sell_value > buy_value * 2:
return {
"bullish": False,
"signal": "STRONG_DISTRIBUTION",
"buy_value": buy_value,
"sell_value": sell_value,
"reason": "Whales are heavily selling"
}
return {
"bullish": buy_value > sell_value,
"signal": "MODERATE",
"buy_value": buy_value,
"sell_value": sell_value,
"reason": "Mixed whale activity"
}
-56
View File
@@ -1,56 +0,0 @@
from loguru import logger
from typing import List, Dict, Optional
from src.data_collection.polymarket_api import PolymarketClient
class OnchainTracker:
"""
追踪 Polymarket 上的大额交易和钱包动向
主要通过 Polymarket API 获取交易历史,并模拟链上分析逻辑
"""
def __init__(self, config: dict, client: PolymarketClient):
self.config = config
self.client = client
self.whale_threshold = self.config.get("whale_threshold", 5000) # $5000 以上视为鲸鱼
logger.info(f"Initializing Onchain Tracker (Whale Threshold: ${self.whale_threshold})")
def get_large_transactions(self, market_id: str, limit: int = 100) -> List[Dict]:
"""
获取特定市场的历史大额交易
"""
trades = self.client.get_trades(market_id=market_id, limit=limit)
if not trades:
return []
# 过滤大额交易 (Polymarket API 返回的格式可能需要根据实际调整)
# 假设格式: [{"price": 0.9, "size": 10000, "side": "BUY", "maker": "0x...", "taker": "0x..."}]
large_trades = []
for trade in trades:
size = float(trade.get("size", 0))
price = float(trade.get("price", 0))
value = size * price
if value >= self.whale_threshold:
trade["value"] = value
large_trades.append(trade)
return large_trades
def get_whale_positions(self, market_id: str) -> Dict[str, float]:
"""
估算大户在某个市场的持仓情况
注意:这只是基于最近交易的估算,真实持仓需要查询链上合约
"""
trades = self.get_large_transactions(market_id, limit=500)
whale_holdings = {}
for trade in trades:
wallet = trade.get("proxyWallet") or trade.get("maker") or "unknown"
side = trade.get("side", "").upper()
size = float(trade.get("size", 0))
if side == "BUY":
whale_holdings[wallet] = whale_holdings.get(wallet, 0) + size
else:
whale_holdings[wallet] = whale_holdings.get(wallet, 0) - size
return whale_holdings
-291
View File
@@ -1,291 +0,0 @@
import os
import requests
import time
import re
from typing import Dict, List, Optional
from loguru import logger
from datetime import datetime
from concurrent.futures import ThreadPoolExecutor
class PolymarketClient:
"""
Polymarket API Client (Pure REST API version)
Directly uses Gamma API and CLOB REST API without py-clob-client dependency.
"""
def __init__(self, config: Dict):
self.clob_url = config.get("base_url", "https://clob.polymarket.com")
self.gamma_url = "https://gamma-api.polymarket.com"
self.timeout = config.get("timeout", 20)
self.session = requests.Session()
# Cache mechanism
self._weather_markets_cache = []
self._last_discovery_time = 0
self._cache_ttl = 300 # 5 minutes cache
# Proxy settings (automatically read from environment)
proxy = os.getenv("HTTPS_PROXY") or os.getenv("HTTP_PROXY")
if proxy:
self.session.proxies = {"http": proxy, "https": proxy}
logger.info(f"Requests session using proxy: {proxy}")
# Set common User-Agent and headers
self.session.headers.update(
{
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Accept": "application/json",
"Content-Type": "application/json"
}
)
self.api_key = config.get("api_key")
if self.api_key:
self.session.headers.update({"POLY_API_KEY": self.api_key})
logger.info(f"Polymarket REST Client initialized. CLOB: {self.clob_url}, Gamma: {self.gamma_url}")
def get_markets(self, next_cursor: str = None) -> Optional[Dict]:
"""Fetch markets list via CLOB REST API"""
try:
params = {}
if next_cursor:
params["next_cursor"] = next_cursor
resp = self.session.get(f"{self.clob_url}/markets", params=params, timeout=self.timeout)
return resp.json() if resp.status_code == 200 else None
except Exception as e:
logger.debug(f"get_markets failed: {e}")
return None
def get_market(self, market_id: str) -> Optional[Dict]:
"""Fetch market details via CLOB REST API"""
try:
resp = self.session.get(f"{self.clob_url}/markets/{market_id}", timeout=self.timeout)
return resp.json() if resp.status_code == 200 else None
except Exception as e:
logger.debug(f"get_market failed: {e}")
return None
def get_price(self, token_id: str, side: str = "ask") -> Optional[float]:
"""Fetch real-time price for a token via CLOB REST API"""
try:
# Correct CLOB Mapping:
# 'sell' side price is the ASK (price you pay to BUY)
# 'buy' side price is the BID (price you get to SELL)
clob_side = "sell" if side.lower() in ["ask", "buy"] else "buy"
resp = self.session.get(
f"{self.clob_url}/price",
params={"token_id": token_id, "side": clob_side},
timeout=10
)
data = resp.json()
return float(data.get("price", 0)) if resp.status_code == 200 else None
except Exception as e:
logger.debug(f"get_price failed ({token_id}): {e}")
return None
def get_orderbook(self, token_id: str) -> Optional[Dict]:
"""Fetch orderbook for a token via CLOB REST API"""
try:
resp = self.session.get(
f"{self.clob_url}/book", params={"token_id": token_id}, timeout=10
)
return resp.json() if resp.status_code == 200 else None
except Exception as e:
logger.debug(f"get_orderbook failed ({token_id}): {e}")
return None
def get_buy_prices(self, yes_token_id: str, no_token_id: str) -> Optional[Dict]:
"""Fetch buy prices for both YES and NO tokens"""
try:
# Buy Yes = Ask price of YES token
buy_yes = self.get_price(yes_token_id, "BUY")
# Buy No = Ask price of NO token
buy_no = self.get_price(no_token_id, "BUY")
if buy_yes is not None and buy_no is not None:
return {"buy_yes": buy_yes, "buy_no": buy_no}
except Exception as e:
logger.debug(f"get_buy_prices failed: {e}")
return None
def get_multiple_prices(self, token_requests: List[Dict]) -> Dict[str, float]:
"""Batch fetch prices for multiple tokens using ThreadPoolExecutor"""
if not token_requests:
return {}
all_prices = {}
def robust_float(val):
try: return float(val)
except: return 0.0
def fetch_single(req):
tid = req["token_id"]
side = req.get("side", "ask").lower()
# To get ASK (price to buy), request 'sell' side
# To get BID (price to sell), request 'buy' side
api_side = "sell" if side == "ask" else "buy"
val = self.get_price(tid, api_side)
if val:
return f"{tid}:{side.lower()}", val
return None
with ThreadPoolExecutor(max_workers=5) as executor:
results = list(executor.map(fetch_single, token_requests))
for res in results:
if res:
key, val = res
all_prices[key] = val
return all_prices
def get_midpoint(self, token_id: str) -> Optional[float]:
"""Fetch midpoint price via CLOB REST API"""
try:
resp = self.session.get(f"{self.clob_url}/midpoint", params={"token_id": token_id}, timeout=10)
data = resp.json()
return float(data.get("mid", 0)) if resp.status_code == 200 else None
except:
return None
def discover_weather_markets(self) -> list:
"""Scan Gamma API for all weather-related markets with prioritized search and city targeting"""
# Cache check
current_time = time.time()
if self._weather_markets_cache and (current_time - self._last_discovery_time < self._cache_ttl):
logger.debug(f"Using cached market list ({len(self._weather_markets_cache)} items)")
return self._weather_markets_cache
logger.info("📡 Scanning Polymarket via Gamma API for weather markets...")
all_weather_markets = []
seen_keys = set()
# 1. Target newest markets by query and ID sorting
search_queries = ["highest temperature", "temperature in", "daily weather"]
try:
# Use multiple offsets to find more historical/diverse markets
for offset in [0, 500, 1000]:
for query in search_queries:
logger.debug(f"Searching with query: {query} (offset {offset})")
params = {
"query": query,
"active": "true",
"limit": 500,
"offset": offset,
"order": "id",
"ascending": "false"
}
resp = self.session.get(f"{self.gamma_url}/markets", params=params, timeout=self.timeout)
if resp.status_code == 200:
markets = resp.json()
logger.debug(f"Query '{query}' returned {len(markets)} markets")
for m in markets:
q = m.get("question", "").lower()
slug = m.get("slug", "").lower()
# Filter for weather markets (Broadened)
is_weather = any(k in q or k in slug for k in [
"highest temperature", "highest-temperature",
"temperature in", "temperature-in",
"daily weather", "daily-weather",
"weather", "气温", "温度"
])
if is_weather:
c_id = m.get("conditionId")
t_ids = m.get("clobTokenIds")
active_id = m.get("activeTokenId")
# Robust JSON parsing for clobTokenIds string
if isinstance(t_ids, str) and t_ids.startswith("["):
try:
import json
t_ids = json.loads(t_ids)
except:
pass
# For Neg Risk markets, activeTokenId might be missing in list view
# If we have clobTokenIds, we can work with it
if not t_ids:
continue
if not active_id and isinstance(t_ids, list) and len(t_ids) > 0:
active_id = t_ids[0] # Assume first is YES
if not active_id:
continue
unique_key = f"{c_id}_{active_id}"
if unique_key not in seen_keys:
logger.debug(f"Found weather segment: {q}")
all_weather_markets.append({
"condition_id": c_id,
"question": m.get("question"),
"active_token_id": active_id,
"outcome_index": t_ids.index(active_id) if isinstance(t_ids, list) and active_id in t_ids else 0,
"tokens": t_ids,
"prices": m.get("outcomePrices"),
"event_title": m.get("description", "")[:100],
"slug": m.get("slug"),
"group_id": m.get("negRiskMarketID")
})
seen_keys.add(unique_key)
else:
logger.debug(f"Query '{query}' failed with status {resp.status_code}")
if len(all_weather_markets) > 50:
break
logger.info(f"Discovery complete: Found {len(all_weather_markets)} weather segments.")
self._weather_markets_cache = all_weather_markets
self._last_discovery_time = current_time
return all_weather_markets
except Exception as e:
logger.error(f"Market discovery failed: {e}")
return []
except Exception as e:
logger.error(f"Market discovery failed: {e}")
return []
except Exception as e:
logger.error(f"Market discovery failed: {e}")
return []
def get_weather_markets(self) -> list:
return self.discover_weather_markets()
def find_weather_market(self, city: str, date_str: str = None) -> Optional[Dict]:
markets = self.get_weather_markets()
for m in markets:
# Match against question, title AND slug
content = (str(m.get("question", "")) + str(m.get("event_title", "")) + str(m.get("slug", ""))).lower()
if city.lower() in content:
if date_str:
if date_str.lower() in content: return m
else:
return m
return None
def get_weather_event_markets(self, city: str) -> list:
all_markets = self.get_weather_markets()
return [
m for m in all_markets
if city.lower() in (str(m.get("question", "")) + str(m.get("event_title", "")) + str(m.get("slug", ""))).lower()
]
# --- Trading Stubs (Real trading requires signing, which is disabled in pure REST mode) ---
def create_order(self, *args, **kwargs) -> Optional[Dict]:
logger.warning("create_order: Real trading is disabled in pure REST mode. Please use paper trading.")
return None
def cancel_order(self, *args, **kwargs) -> Optional[Dict]:
logger.warning("cancel_order: Real trading is disabled in pure REST mode.")
return None
def get_orders(self, *args, **kwargs) -> Optional[Dict]:
logger.warning("get_orders: Real trading is disabled in pure REST mode.")
return None
View File
-285
View File
@@ -1,285 +0,0 @@
import numpy as np
import pandas as pd
from typing import Dict, List, Optional, Tuple
from datetime import datetime
from loguru import logger
try:
from statsmodels.tsa.arima.model import ARIMA
HAS_STATSMODELS = True
except ImportError:
HAS_STATSMODELS = False
logger.debug("statsmodels not installed, ARIMA model unavailable")
try:
from sklearn.ensemble import RandomForestRegressor
from sklearn.model_selection import train_test_split
HAS_SKLEARN = True
except ImportError:
HAS_SKLEARN = False
logger.debug("scikit-learn not installed, ML models unavailable")
class TemperaturePredictor:
"""
Temperature prediction model using statistical and ML methods
Supports:
- ARIMA for time series prediction
- Random Forest for feature-based prediction
- Ensemble of both methods
"""
def __init__(self, config: dict = None):
self.config = config or {}
self.arima_order = self.config.get("arima_order", (5, 1, 2))
self.rf_estimators = self.config.get("rf_estimators", 100)
self.arima_model = None
self.rf_model = None
self.is_trained = False
logger.info("Temperature Predictor initialized")
def prepare_features(self, data: pd.DataFrame) -> pd.DataFrame:
"""
Prepare features for ML model
Args:
data: DataFrame with temperature history
Returns:
DataFrame: Feature-engineered data
"""
df = data.copy()
# Time-based features
if 'date' in df.columns:
df['date'] = pd.to_datetime(df['date'])
df['day_of_year'] = df['date'].dt.dayofyear
df['month'] = df['date'].dt.month
df['day_of_week'] = df['date'].dt.dayofweek
# Lag features
if 'temp' in df.columns:
for lag in [1, 2, 3, 7, 14]:
df[f'temp_lag_{lag}'] = df['temp'].shift(lag)
# Rolling statistics
df['temp_rolling_mean_7'] = df['temp'].rolling(window=7).mean()
df['temp_rolling_std_7'] = df['temp'].rolling(window=7).std()
df['temp_rolling_mean_14'] = df['temp'].rolling(window=14).mean()
# Drop NaN rows created by lag features
df = df.dropna()
return df
def train_arima(self, temperature_series: List[float]) -> bool:
"""
Train ARIMA model on temperature time series
Args:
temperature_series: List of historical temperatures
Returns:
bool: Success status
"""
if not HAS_STATSMODELS:
logger.error("statsmodels required for ARIMA training")
return False
if len(temperature_series) < 30:
logger.warning("Insufficient data for ARIMA training (need 30+ points)")
return False
try:
series = np.array(temperature_series)
model = ARIMA(series, order=self.arima_order)
self.arima_model = model.fit()
logger.info(f"ARIMA model trained. AIC: {self.arima_model.aic:.2f}")
return True
except Exception as e:
logger.error(f"ARIMA training failed: {e}")
return False
def train_random_forest(self,
features: pd.DataFrame,
target_col: str = 'temp') -> bool:
"""
Train Random Forest model
Args:
features: Feature DataFrame
target_col: Target column name
Returns:
bool: Success status
"""
if not HAS_SKLEARN:
logger.error("scikit-learn required for Random Forest training")
return False
if len(features) < 50:
logger.warning("Insufficient data for RF training (need 50+ rows)")
return False
try:
# Prepare data
feature_cols = [c for c in features.columns
if c not in [target_col, 'date', 'datetime']]
X = features[feature_cols].values
y = features[target_col].values
# Train-test split
X_train, X_test, y_train, y_test = train_test_split(
X, y, test_size=0.2, random_state=42
)
# Train model
self.rf_model = RandomForestRegressor(
n_estimators=self.rf_estimators,
random_state=42,
n_jobs=-1
)
self.rf_model.fit(X_train, y_train)
# Evaluate
train_score = self.rf_model.score(X_train, y_train)
test_score = self.rf_model.score(X_test, y_test)
logger.info(f"Random Forest trained. Train R²: {train_score:.4f}, Test R²: {test_score:.4f}")
# Store feature names
self.feature_names = feature_cols
return True
except Exception as e:
logger.error(f"Random Forest training failed: {e}")
return False
def predict_arima(self, steps: int = 1) -> Optional[Dict]:
"""
Make prediction using ARIMA model
Args:
steps: Number of steps to forecast
Returns:
dict: Prediction with confidence interval
"""
if self.arima_model is None:
logger.warning("ARIMA model not trained")
return None
try:
forecast = self.arima_model.forecast(steps=steps)
conf_int = self.arima_model.get_forecast(steps=steps).conf_int()
return {
"method": "ARIMA",
"predicted_temp": float(forecast[0]) if steps == 1 else [float(f) for f in forecast],
"confidence_interval": [float(conf_int.iloc[0, 0]), float(conf_int.iloc[0, 1])] if steps == 1 else conf_int.values.tolist()
}
except Exception as e:
logger.error(f"ARIMA prediction failed: {e}")
return None
def predict_rf(self, features: np.ndarray) -> Optional[Dict]:
"""
Make prediction using Random Forest model
Args:
features: Feature array for prediction
Returns:
dict: Prediction result
"""
if self.rf_model is None:
logger.warning("Random Forest model not trained")
return None
try:
prediction = self.rf_model.predict(features.reshape(1, -1))[0]
# Estimate confidence using tree variance
tree_predictions = [tree.predict(features.reshape(1, -1))[0]
for tree in self.rf_model.estimators_]
std = np.std(tree_predictions)
return {
"method": "RandomForest",
"predicted_temp": float(prediction),
"confidence_interval": [float(prediction - 1.96 * std),
float(prediction + 1.96 * std)],
"std": float(std)
}
except Exception as e:
logger.error(f"Random Forest prediction failed: {e}")
return None
def predict_ensemble(self,
temperature_history: List[float],
feature_data: pd.DataFrame = None,
arima_weight: float = 0.4,
rf_weight: float = 0.6) -> Dict:
"""
Make ensemble prediction combining ARIMA and Random Forest
Args:
temperature_history: Historical temperature series
feature_data: Feature data for RF prediction
arima_weight: Weight for ARIMA prediction
rf_weight: Weight for RF prediction
Returns:
dict: Ensemble prediction
"""
predictions = []
weights = []
# ARIMA prediction
if self.arima_model is not None:
arima_pred = self.predict_arima(steps=1)
if arima_pred:
predictions.append(arima_pred["predicted_temp"])
weights.append(arima_weight)
# Random Forest prediction
if self.rf_model is not None and feature_data is not None:
# Get latest features
prepared = self.prepare_features(feature_data)
if len(prepared) > 0 and hasattr(self, 'feature_names'):
latest_features = prepared[self.feature_names].iloc[-1].values
rf_pred = self.predict_rf(latest_features)
if rf_pred:
predictions.append(rf_pred["predicted_temp"])
weights.append(rf_weight)
if not predictions:
logger.debug("No predictions available (Model not trained)")
return {
"predicted_temp": None,
"confidence": 0.5,
"error": "No models available for prediction"
}
# Weighted average
weights = np.array(weights) / np.sum(weights) # Normalize weights
ensemble_pred = np.average(predictions, weights=weights)
# Estimate confidence based on model agreement
if len(predictions) > 1:
spread = abs(predictions[0] - predictions[1])
confidence = max(0.5, 1.0 - spread / 5.0) # Lower confidence if predictions differ
else:
confidence = 0.7
return {
"predicted_temp": float(ensemble_pred),
"confidence": confidence,
"confidence_interval": [ensemble_pred - 2.0, ensemble_pred + 2.0], # Approximate
"individual_predictions": predictions,
"weights": weights.tolist()
}
View File
-198
View File
@@ -1,198 +0,0 @@
from loguru import logger
from src.analysis.volume_analyzer import VolumeAnalyzer
from src.analysis.orderbook_analyzer import analyze_orderbook
from src.analysis.technical_indicators import TechnicalIndicators
class DecisionEngine:
"""
综合决策引擎 - 多因子加权评分系统
"""
def __init__(self, config: dict = None):
self.config = config or {}
# 因子权重
self.weights = self.config.get("weights", {
"statistical_prediction": 0.50,
"data_source_consensus": 0.15,
"market_volume_signal": 0.15,
"orderbook_analysis": 0.10,
"technical_indicators": 0.05,
"onchain_whale_signal": 0.05
})
# 初始化分析器
self.volume_analyzer = VolumeAnalyzer(config)
self.tech_indicators = TechnicalIndicators()
logger.info("决策引擎初始化完成。")
logger.debug(f"权重配置: {self.weights}")
def calculate_signal(self,
model_prediction: dict,
market_data: dict,
weather_consensus: dict = None,
whale_activity: dict = None) -> dict:
"""
综合多因子计算交易信号
Args:
model_prediction: 统计模型预测结果
market_data: 市场数据 (价格历史、订单簿、交易量等)
weather_consensus: 天气数据源一致性检查结果
whale_activity: 链上大户活动数据
Returns:
dict: 综合评分和交易建议
"""
scores = {}
details = {}
# 1. 统计模型预测得分 (权重: 50%)
stat_confidence = model_prediction.get("confidence", 0.5)
scores["statistical"] = stat_confidence
details["statistical"] = {
"score": stat_confidence,
"prediction": model_prediction.get("predicted_temp"),
"confidence_interval": model_prediction.get("confidence_interval")
}
# 2. 多源数据一致性 (权重: 15%)
if weather_consensus:
is_consensus = weather_consensus.get("consensus", False)
consensus_score = 1.0 if is_consensus else 0.3
else:
consensus_score = 0.5
scores["consensus"] = consensus_score
details["consensus"] = weather_consensus
# 3. 交易量信号 (权重: 15%)
volume_history = market_data.get("volume_history", [])
transactions = market_data.get("transactions", [])
volume_analysis = self.volume_analyzer.analyze(volume_history, transactions)
scores["volume"] = volume_analysis.get("combined_score", 0.5)
details["volume"] = volume_analysis
# 4. 订单簿分析 (权重: 10%)
orderbook = market_data.get("orderbook", {})
orderbook_signal = analyze_orderbook(orderbook)
scores["orderbook"] = orderbook_signal.get("confidence", 0.5)
details["orderbook"] = orderbook_signal
# 5. 技术指标 (权重: 5%)
price_history = market_data.get("price_history", [])
if price_history:
tech_signal = self.tech_indicators.get_signal(price_history)
scores["technical"] = tech_signal.get("combined_score", 0.5)
details["technical"] = tech_signal
else:
scores["technical"] = 0.5
details["technical"] = {"message": "No price history available"}
# 6. 链上鲸鱼信号 (权重: 5%)
if whale_activity:
is_bullish = whale_activity.get("bullish", False)
whale_score = 0.8 if is_bullish else 0.2
else:
whale_score = 0.5
scores["whale"] = whale_score
details["whale"] = whale_activity
# 加权计算最终分数
final_score = (
scores["statistical"] * self.weights["statistical_prediction"] +
scores["consensus"] * self.weights["data_source_consensus"] +
scores["volume"] * self.weights["market_volume_signal"] +
scores["orderbook"] * self.weights["orderbook_analysis"] +
scores["technical"] * self.weights["technical_indicators"] +
scores["whale"] * self.weights["onchain_whale_signal"]
)
# 生成建议
recommendation = self._get_recommendation(final_score)
result = {
"final_score": round(final_score, 4),
"recommendation": recommendation,
"factor_scores": scores,
"factor_details": details,
"weights": self.weights
}
logger.info(f"Decision: {recommendation} (score: {final_score:.4f})")
return result
def _get_recommendation(self, score: float) -> str:
"""
根据评分生成交易建议
Args:
score: 综合评分 (0-1)
Returns:
str: 交易建议
"""
if score > 0.80:
return "STRONG_BUY"
elif score > 0.65:
return "BUY"
elif score > 0.50:
return "WEAK_BUY"
elif score > 0.35:
return "HOLD"
elif score > 0.20:
return "WEAK_SELL"
else:
return "NO_ACTION"
def should_trade(self,
signal: dict,
current_price: float,
min_confidence: float = 0.65) -> dict:
"""
判断是否应该执行交易
Args:
signal: calculate_signal返回的信号
current_price: 当前市场价格
min_confidence: 最低置信度阈值
Returns:
dict: 交易决策
"""
final_score = signal.get("final_score", 0)
recommendation = signal.get("recommendation", "NO_ACTION")
# 检查是否满足交易条件
should_buy = (
final_score >= min_confidence and
recommendation in ["STRONG_BUY", "BUY"] and
current_price >= 0.85 # 价格阈值
)
should_sell = (
final_score < 0.35 or
recommendation in ["WEAK_SELL", "NO_ACTION"]
)
if should_buy:
return {
"action": "BUY",
"confidence": final_score,
"price": current_price,
"reason": f"Score {final_score:.2f} >= threshold {min_confidence}"
}
elif should_sell:
return {
"action": "SELL",
"confidence": final_score,
"price": current_price,
"reason": f"Score {final_score:.2f} below threshold or bearish signal"
}
else:
return {
"action": "HOLD",
"confidence": final_score,
"price": current_price,
"reason": "Conditions not met for trading"
}
-153
View File
@@ -1,153 +0,0 @@
from loguru import logger
class PositionManager:
"""
仓位管理 - Kelly公式动态仓位计算
"""
def __init__(self, config=None):
self.config = config or {}
self.max_position_ratio = self.config.get("max_position_ratio", 0.25) # 最大单笔25%
self.max_total_exposure = self.config.get("max_total_exposure", 0.80) # 最大总仓位80%
self.min_trade_size = self.config.get("min_trade_size", 10) # 最小交易额$10
logger.info("Initializing Position Manager...")
def kelly_criterion(self, win_prob: float, odds: float) -> float:
"""
Kelly公式计算最优投资比例
f = (bp - q) / b
f: 应投资的资金比例
b: 赔率 (盈利/亏损)
p: 胜率
q: 败率 (1-p)
Args:
win_prob: 预测胜率 (0-1)
odds: 赔率
Returns:
float: 建议投资比例 (0-1)
"""
if win_prob <= 0 or win_prob >= 1 or odds <= 0:
return 0.0
q = 1 - win_prob
f = (win_prob * odds - q) / odds
# 限制最大仓位
f = max(0, min(f, self.max_position_ratio))
logger.debug(f"Kelly ratio: {f:.4f} (win_prob={win_prob:.2f}, odds={odds:.2f})")
return f
def calculate_position_size(self,
total_capital: float,
win_prob: float,
market_price: float,
current_exposure: float = 0) -> dict:
"""
计算建议仓位大小
Args:
total_capital: 总资金
win_prob: 模型预测胜率
market_price: 当前市场价格 (0-1)
current_exposure: 当前已有仓位占比
Returns:
dict: 包含建议仓位大小和相关信息
"""
# 计算赔率
if market_price <= 0 or market_price >= 1:
return {"size": 0, "error": "Invalid market price"}
odds = (1 - market_price) / market_price
# Kelly计算
kelly_ratio = self.kelly_criterion(win_prob, odds)
# 检查总仓位限制
available_ratio = self.max_total_exposure - current_exposure
if available_ratio <= 0:
return {
"size": 0,
"kelly_ratio": kelly_ratio,
"reason": "Max exposure reached"
}
# 实际使用比例
actual_ratio = min(kelly_ratio, available_ratio)
# 计算金额
position_size = total_capital * actual_ratio
# 检查最小交易额
if position_size < self.min_trade_size:
return {
"size": 0,
"kelly_ratio": kelly_ratio,
"reason": f"Below minimum trade size (${self.min_trade_size})"
}
return {
"size": position_size,
"kelly_ratio": kelly_ratio,
"actual_ratio": actual_ratio,
"odds": odds,
"expected_return": (win_prob * odds - (1 - win_prob)) * position_size
}
def should_exit(self,
entry_price: float,
current_price: float,
current_prediction: float,
stop_loss: float = 0.15,
take_profit: float = 0.30) -> dict:
"""
判断是否应该平仓
Args:
entry_price: 入场价格
current_price: 当前价格
current_prediction: 当前模型预测
stop_loss: 止损比例
take_profit: 止盈比例
Returns:
dict: 退出建议
"""
if entry_price <= 0:
return {"should_exit": False}
pnl_ratio = (current_price - entry_price) / entry_price
# 止损
if pnl_ratio < -stop_loss:
return {
"should_exit": True,
"reason": "STOP_LOSS",
"pnl_ratio": pnl_ratio
}
# 止盈
if pnl_ratio > take_profit:
return {
"should_exit": True,
"reason": "TAKE_PROFIT",
"pnl_ratio": pnl_ratio
}
# 模型预测反转
if current_prediction < 0.4: # 预测胜率下降
return {
"should_exit": True,
"reason": "PREDICTION_REVERSAL",
"pnl_ratio": pnl_ratio,
"current_prediction": current_prediction
}
return {
"should_exit": False,
"pnl_ratio": pnl_ratio
}
-99
View File
@@ -1,99 +0,0 @@
from loguru import logger
class RiskManager:
"""
风险控制系统
"""
def __init__(self, config=None):
self.config = config or {}
# 基础风控参数
self.max_single_trade = self.config.get(
"max_single_trade", 50.0
) # 最大单笔调整为 $50
self.max_daily_exposure = 50.0 # 每日最高投入上限
self.daily_used_exposure = 0.0
self.last_reset_date = ""
self.min_confidence = 0.5
self.peak_capital = 0
self.is_trading_paused = False
logger.info("Initializing Pro Risk Manager...")
def _reset_daily_exposure(self):
"""每日重置额度"""
from datetime import datetime
today = datetime.now().strftime("%Y-%m-%d")
if self.last_reset_date != today:
self.daily_used_exposure = 0.0
self.last_reset_date = today
logger.info(f"Daily exposure reset for {today}")
def calculate_position_size(
self,
base_confidence_usd: float,
depth: float = 0,
hours_to_settle: float = 24,
is_high_relative_volume: bool = False,
) -> tuple[float, str]:
"""
仓位计算方法 (简化版移除流动性过滤):
仓位 = base_position(置信度)
× time_decay(离结算衰减)
× budget_limit
"""
self._reset_daily_exposure()
final_pos = base_confidence_usd
reason = "Normal"
# 1. 时间衰减因子
# 离结算时间越近,预测越准但也存在剧烈博弈风险
time_factor = 1.0
if hours_to_settle <= 1.0:
time_factor = 0.0 # 最后 1 小时停止建仓
reason = "🚫临近结算"
elif hours_to_settle <= 4.0:
time_factor = 0.4 # 1-4小时:缩小 60%
reason = "⏱️结算冲刺 (40%)"
elif hours_to_settle <= 12.0:
time_factor = 0.7 # 4-12小时:缩小 30%
reason = "⏳接近结算 (70%)"
final_pos *= time_factor
if final_pos <= 0:
return 0.0, reason
# 2. 预算上限过滤
remaining_daily = self.max_daily_exposure - self.daily_used_exposure
if remaining_daily <= 0:
return 0.0, "🚫今日总额度已满 ($50)"
if final_pos > remaining_daily:
final_pos = remaining_daily
reason = "🛑触及日风控上限"
# 3. 高相对成交量加权 (如果是高成交量市场,且逻辑支持,可保持原状或微增)
# 这里逻辑设定为:如果不是高成交量,再次缩减 20% 防御
if not is_high_relative_volume:
final_pos *= 0.8
if reason == "Normal":
reason = "📉低活缩减"
return round(final_pos, 2), reason
def record_trade(self, amount: float):
"""记录成交额以扣除额度"""
self.daily_used_exposure += amount
logger.debug(
f"Applied exposure: ${amount}. Daily Total: ${self.daily_used_exposure}"
)
def check_trade_risk(
self, trade_size: float, market_data: dict, model_confidence: float
) -> dict:
"""保持基础接口兼容"""
return {"passed": True, "risks": []}
View File
-219
View File
@@ -1,219 +0,0 @@
from loguru import logger
from typing import Optional, Dict
from src.data_collection.polymarket_api import PolymarketClient
class OrderExecutor:
"""
交易执行器 - 负责订单生成提交和管理
"""
def __init__(self, config: dict, client: PolymarketClient):
self.config = config
self.client = client
self.pending_orders = {}
self.executed_orders = []
logger.info("Order Executor initialized")
def execute_trade(self,
token_id: str,
side: str,
amount: float,
price: float,
order_type: str = "GTC") -> Dict:
"""
执行交易
Args:
token_id: Token ID
side: "BUY" "SELL"
amount: 交易金额
price: 价格
order_type: 订单类型 (GTC, GTD, FOK)
Returns:
dict: 订单结果
"""
logger.info(f"Executing {side} order: ${amount:.2f} @ {price:.4f}")
# 计算数量
if price <= 0:
return {"status": "error", "message": "Invalid price"}
size = amount / price
# 提交订单
try:
result = self.client.create_order(
token_id=token_id,
side=side,
price=price,
size=size,
order_type=order_type
)
if result:
order_id = result.get("orderID", "unknown")
self.executed_orders.append({
"order_id": order_id,
"token_id": token_id,
"side": side,
"price": price,
"size": size,
"amount": amount,
"result": result
})
logger.info(f"Order executed successfully: {order_id}")
return {
"status": "success",
"order_id": order_id,
"side": side,
"price": price,
"size": size,
"amount": amount
}
else:
return {"status": "error", "message": "Order submission failed"}
except Exception as e:
logger.error(f"Order execution failed: {e}")
return {"status": "error", "message": str(e)}
def cancel_order(self, order_id: str) -> Dict:
"""
取消订单
Args:
order_id: 订单ID
Returns:
dict: 取消结果
"""
try:
result = self.client.cancel_order(order_id)
if result:
logger.info(f"Order {order_id} cancelled")
return {"status": "success", "order_id": order_id}
else:
return {"status": "error", "message": "Cancel failed"}
except Exception as e:
logger.error(f"Cancel order failed: {e}")
return {"status": "error", "message": str(e)}
def get_open_orders(self, market_id: str = None) -> Optional[Dict]:
"""
获取当前挂单
Args:
market_id: 可选的市场过滤
Returns:
dict: 挂单列表
"""
return self.client.get_orders(market_id)
def get_execution_history(self) -> list:
"""
获取执行历史
Returns:
list: 已执行订单列表
"""
return self.executed_orders
class PortfolioTracker:
"""
持仓追踪器
"""
def __init__(self):
self.positions = {}
self.total_invested = 0
self.total_pnl = 0
logger.info("Portfolio Tracker initialized")
def add_position(self,
token_id: str,
side: str,
size: float,
entry_price: float,
amount: float):
"""
添加持仓
"""
if token_id not in self.positions:
self.positions[token_id] = {
"side": side,
"size": size,
"entry_price": entry_price,
"amount": amount,
"current_price": entry_price,
"unrealized_pnl": 0
}
else:
# 加仓
existing = self.positions[token_id]
total_size = existing["size"] + size
avg_price = (existing["size"] * existing["entry_price"] + size * entry_price) / total_size
existing["size"] = total_size
existing["entry_price"] = avg_price
existing["amount"] += amount
self.total_invested += amount
logger.info(f"Position added: {token_id}, size={size}, price={entry_price}")
def update_price(self, token_id: str, current_price: float):
"""
更新持仓价格
"""
if token_id in self.positions:
pos = self.positions[token_id]
pos["current_price"] = current_price
# 计算未实现盈亏
if pos["side"] == "BUY":
pos["unrealized_pnl"] = (current_price - pos["entry_price"]) * pos["size"]
else:
pos["unrealized_pnl"] = (pos["entry_price"] - current_price) * pos["size"]
def close_position(self, token_id: str, exit_price: float) -> Dict:
"""
平仓
"""
if token_id not in self.positions:
return {"status": "error", "message": "Position not found"}
pos = self.positions[token_id]
if pos["side"] == "BUY":
realized_pnl = (exit_price - pos["entry_price"]) * pos["size"]
else:
realized_pnl = (pos["entry_price"] - exit_price) * pos["size"]
self.total_pnl += realized_pnl
self.total_invested -= pos["amount"]
del self.positions[token_id]
return {
"status": "success",
"realized_pnl": realized_pnl,
"exit_price": exit_price
}
def get_summary(self) -> Dict:
"""
获取持仓汇总
"""
total_unrealized = sum(p["unrealized_pnl"] for p in self.positions.values())
return {
"positions_count": len(self.positions),
"total_invested": self.total_invested,
"total_unrealized_pnl": total_unrealized,
"total_realized_pnl": self.total_pnl,
"positions": self.positions
}
-158
View File
@@ -1,158 +0,0 @@
import json
import os
import time
from datetime import datetime, timedelta
from loguru import logger
class PaperTrader:
"""
模拟交易系统 (Paper Trading System)
"""
def __init__(self, storage_path="data/paper_positions.json", total_capital=1000.0):
self.storage_path = storage_path
self.initial_capital = total_capital
data = self._load_data()
self.positions = data.get("positions", {})
self.history = data.get("history", []) # 历史结项记录
self.trades = data.get("trades", []) # 原始买入/卖出记录
self.balance = data.get("balance", total_capital)
logger.info(f"模拟交易系统初始化。累计成交: {len(self.history)} 笔, 买入记录: {len(self.trades)}")
def _load_data(self):
if os.path.exists(self.storage_path):
try:
with open(self.storage_path, "r", encoding="utf-8") as f:
return json.load(f)
except:
return {"positions": {}, "history": [], "trades": [], "balance": self.initial_capital}
return {"positions": {}, "history": [], "trades": [], "balance": self.initial_capital}
def _save_data(self):
with open(self.storage_path, "w", encoding="utf-8") as f:
json.dump(
{
"positions": self.positions,
"history": self.history,
"trades": self.trades,
"balance": round(self.balance, 2),
},
f,
ensure_ascii=False,
indent=2,
)
def open_position(self, market_id: str, city: str, option: str, price: int, side: str, amount_usd: float = 5.0, target_date: str = None, predicted_temp: float = None):
"""
开仓进入模拟仓位
"""
# 价格以美分计,转换为 0-1 比例
price_decimal = price / 100.0
# 检查余额
if self.balance < amount_usd:
logger.warning(f"余额不足,无法开仓 (余额: ${self.balance:.2f})")
return False
# 计算持仓份额
shares = amount_usd / price_decimal if price_decimal > 0 else 0
position_id = f"{market_id}_{side}"
# 如果已经有相同方向的仓位,可以选择加仓或忽略(这里简单起见,不重复开仓)
if position_id in self.positions:
return False
new_pos = {
"market_id": market_id,
"city": city,
"option": option,
"side": side,
"entry_price": price,
"shares": shares,
"cost_usd": amount_usd,
"current_price": price,
"pnl_usd": 0.0,
"pnl_pct": 0.0,
"status": "OPEN",
"target_date": target_date,
"predicted_temp": predicted_temp,
"opened_at": (datetime.utcnow() + timedelta(hours=8)).strftime("%Y-%m-%d %H:%M:%S")
}
self.positions[position_id] = new_pos
self.balance -= amount_usd
# 记录交易流水
self.trades.append({
"type": "BUY",
"city": city,
"option": option,
"side": side,
"price": price,
"amount": amount_usd,
"time": new_pos["opened_at"]
})
self._save_data()
logger.success(f"【模拟开仓】{city} | {option} | {side} | 价格: {price}¢ | 投入: ${amount_usd}")
return True
def update_pnl(self, current_prices: dict):
updated_report = []
finished_ids = []
for pid, pos in self.positions.items():
if pos["status"] != "OPEN":
continue
m_id = pos["market_id"]
if m_id in current_prices:
curr_price = current_prices[m_id].get("price", 50)
if pos["side"] == "NO":
curr_price = 100 - curr_price
# 更新当前价值
value = pos["shares"] * (curr_price / 100.0)
pnl = value - pos["cost_usd"]
pnl_pct = (pnl / pos["cost_usd"]) * 100 if pos["cost_usd"] > 0 else 0
pos["current_price"] = curr_price
pos["pnl_usd"] = round(pnl, 2)
pos["pnl_pct"] = round(pnl_pct, 2)
# --- 自动结项检测:如果价格变为 0 或 100 (Polymarket 已结算) ---
if curr_price >= 99.5 or curr_price <= 0.5:
pos["status"] = "CLOSED"
pos["closed_at"] = (
datetime.utcnow() + timedelta(hours=8)
).strftime("%Y-%m-%d %H:%M:%S")
self.balance += value # 资金回笼
self.history.append(pos)
finished_ids.append(pid)
logger.success(
f"【模拟结项】{pos['city']} | {pos['option']} | 最终价格: {curr_price}¢ | 获利: ${pnl:+.2f}"
)
else:
updated_report.append(pos)
# 从活跃仓位中移除已结项的
for pid in finished_ids:
# 在流水中添加卖出(结项)记录
pos = self.positions[pid]
self.trades.append({
"type": "SELL",
"city": pos["city"],
"option": pos["option"],
"side": pos["side"],
"price": pos["current_price"],
"amount": round(pos["shares"] * (pos["current_price"] / 100.0), 2),
"time": pos.get("closed_at")
})
del self.positions[pid]
self._save_data()
return updated_report
-285
View File
@@ -1,285 +0,0 @@
import requests
import html
from loguru import logger
from datetime import datetime
class TelegramNotifier:
"""
Telegram 消息推送模块
支持信号推送预警推送和市场异常提醒
"""
def __init__(self, config: dict):
self.config = config
self.token = config.get("bot_token")
self.chat_id = config.get("chat_id")
self.proxy = config.get("proxy")
self.session = requests.Session()
if self.proxy:
if not self.proxy.startswith("http"):
self.proxy = f"http://{self.proxy}"
self.session.proxies = {"http": self.proxy, "https": self.proxy}
logger.info("Telegram 通知器初始化完成。")
@staticmethod
def _escape_html(text: str) -> str:
"""Escape HTML special characters"""
if not isinstance(text, str):
text = str(text)
return html.escape(text, quote=False)
def _send_message(self, text: str):
"""发送 Telegram 消息的主函数 (支持多个 ID)"""
if not self.token or not self.chat_id:
logger.warning("未配置 Telegram Token 或 Chat ID,无法发送消息。")
return False
# 支持逗号分隔的多个 ID
chat_ids = str(self.chat_id).replace(" ", "").split(",")
url = f"https://api.telegram.org/bot{self.token}/sendMessage"
all_successful = True
for cid in chat_ids:
if not cid:
continue
payload = {
"chat_id": cid,
"text": text,
"parse_mode": "HTML",
"disable_web_page_preview": True,
}
try:
response = self.session.post(url, json=payload, timeout=10)
if response.status_code != 200:
error_msg = response.text
if "chat not found" in error_msg.lower():
logger.error(
f"Telegram 消息发送给 {cid} 失败 (400): Chat ID {cid} 无效或机器人尚未被加入该聊天。请在 Telegram 中发送 /id 给机器人确认正确的 Chat ID。"
)
else:
logger.error(
f"Telegram 消息发送给 {cid} 失败 ({response.status_code}): {error_msg}"
)
all_successful = False
else:
logger.info(f"Telegram 消息发送给 {cid} 成功。")
except Exception as e:
logger.error(f"Telegram 消息发送给 {cid} 异常: {e}")
all_successful = False
return all_successful
def send_signal(
self,
market_name: str,
full_title: str,
option: str,
score: float,
prediction: str,
confidence: int,
analysis_list: list,
price: float,
market_url: str,
local_time: str = None,
target_date: str = None,
):
"""发送交易信号推送"""
stars = "" * int(score) + "" * (5 - int(score))
timestamp_utc = datetime.utcnow().strftime("%H:%M")
analysis_text = "\n".join(
[
f"{self._escape_html(item)}" if "" not in item else item
for item in analysis_list
]
)
local_time_text = (
f"🕒 当地时间: <b>{self._escape_html(local_time)}</b>\n"
if local_time
else ""
)
target_date_text = self._escape_html(target_date) if target_date else "待定"
text = (
f"🎯 <b>交易信号 #{self._escape_html(market_name.split(' ')[0])}</b>\n\n"
f"📍 城市: <b>{self._escape_html(market_name)}</b>\n"
f"🏆 市场: <i>{self._escape_html(full_title)}</i>\n"
f"📝 选项: <b>{self._escape_html(option)}</b>\n"
f"💰 当前价格: <b>{price}¢</b>\n"
f"═══════════════════\n"
f"📊 信号评分: {stars} ({score}/5)\n"
f"🤖 模型预测: {self._escape_html(prediction)}\n"
f"📈 置信度: {confidence}%\n\n"
f"分析汇总:\n"
f"{analysis_text}\n"
f"═══════════════════\n"
f"{local_time_text}"
f"📅 结算日期: <b>{target_date_text}</b>\n"
f"🔗 <a href='{market_url}'>点击进入市场</a>\n\n"
f"⏰ 信号时间: {timestamp_utc} UTC"
)
return self._send_message(text)
def send_combined_alert(
self,
city: str,
alerts: list,
local_time: str = None,
forecast_temp: str = None,
total_volume: float = 0,
brackets_count: int = 0,
strategy_tips: list = None,
metar_data: dict = None,
):
"""发送简约版合并预警 (含 METAR 航空气象数据)"""
if not alerts:
return
from datetime import datetime, timedelta
# UTC+8 北京时间
now_bj = datetime.utcnow() + timedelta(hours=8)
timestamp_bj = now_bj.strftime("%H:%M")
# 1. METAR 航空气象数据区块
metar_text = ""
if metar_data and metar_data.get("current", {}).get("temp") is not None:
icao = metar_data.get("icao", "N/A")
temp = metar_data["current"]["temp"]
unit = "°F" if metar_data.get("unit") == "fahrenheit" else "°C"
# 解析观测时间 (格式: 2026-02-07T11:00:00.000Z)
obs_time_raw = metar_data.get("observation_time", "")
if "T" in obs_time_raw:
obs_time = obs_time_raw.split("T")[1][:5] + " UTC"
else:
obs_time = obs_time_raw or "N/A"
# 可选:风速信息
wind_kt = metar_data["current"].get("wind_speed_kt")
wind_text = f" | 风速:{wind_kt}kt" if wind_kt else ""
metar_text = (
f"✈️ <b>机场实测 ({icao}):</b>\n"
f" 🌡️ {temp:.1f}{unit}{wind_text}\n"
f" 🕐 观测: {obs_time}\n\n"
)
# 2. 信号详情构建
items_text = ""
for a in alerts:
items_text += f"{a['msg']}\n\n"
# 3. 策略建议(如果有)
tips_text = ""
if strategy_tips:
tips_text = (
"💡 <b>策略建议:</b>\n"
+ "\n".join([f"{self._escape_html(tip)}" for tip in strategy_tips])
+ "\n\n"
)
# 4. 总体布局
text = (
f"🔔 <b>城市监控报告 #{self._escape_html(city)}</b>\n\n"
f"📍 城市: {self._escape_html(city)}\n"
f"{metar_text}"
f"📊 <b>实时异动:</b>\n"
f"{items_text}"
f"{tips_text}"
f"═══════════════════\n"
f"🕒 当地时间: {self._escape_html(local_time or 'N/A')}\n"
f"⏰ 预警时间: {timestamp_bj} (北京时间)"
)
return self._send_message(text)
def send_anomaly(
self,
city_tag: str,
market_name: str,
detected_anomaly: str,
stats: dict,
whales: list,
current_price: float,
local_time: str = None,
):
"""发送市场异常推送"""
from datetime import datetime, timedelta
# UTC+8 北京时间
timestamp_bj = (datetime.utcnow() + timedelta(hours=8)).strftime("%H:%M")
whale_text = "\n".join([f"- {self._escape_html(w)}" for w in whales])
stats_text = "\n".join(
[
f"{self._escape_html(k)}: {self._escape_html(v)}"
for k, v in stats.items()
]
)
local_time_text = (
f"🕒 当地时间: <b>{self._escape_html(local_time)}</b>\n"
if local_time
else ""
)
text = (
f"👀 <b>市场异常 #{self._escape_html(city_tag)}</b>\n\n"
f"📍 城市: {self._escape_html(city_tag)}\n"
f"🏆 市场: {self._escape_html(market_name)}\n\n"
f"🚨 <b>检测到异常:</b>\n"
f"{self._escape_html(detected_anomaly)}\n"
f"{stats_text}\n\n"
f"🐋 <b>大户动向:</b>\n"
f"{whale_text}\n\n"
f"💰 当前价格: <b>{current_price}¢</b>\n"
f"═══════════════════\n"
f"{local_time_text}"
f"⏰ 信号时间: {timestamp_bj} (北京时间)"
)
return self._send_message(text)
def send_alert(
self,
city_tag: str,
market_name: str,
price: float,
trigger: str,
prev_price: float,
change: str,
quick_analysis: list,
local_time: str = None,
):
"""发送价格预警推送"""
from datetime import datetime, timedelta
# UTC+8 北京时间
timestamp_bj = (datetime.utcnow() + timedelta(hours=8)).strftime("%H:%M")
analysis_text = "\n".join(
[f"- {self._escape_html(item)}" for item in quick_analysis]
)
local_time_text = (
f"🕒 当地时间: <b>{self._escape_html(local_time)}</b>\n"
if local_time
else ""
)
text = (
f"⚡ <b>价格预警 #{self._escape_html(city_tag)}</b>\n\n"
f"📍 城市: {self._escape_html(city_tag)}\n"
f"🏆 市场: {self._escape_html(market_name)}\n"
f"💰 报价: <b>{price}¢ ↗️</b>\n\n"
f"触发条件: {self._escape_html(trigger)}\n"
f"变动详情: {prev_price}¢ -> {price}¢ ({self._escape_html(change)})\n\n"
f"📊 <b>快速分析:</b>\n"
f"{analysis_text}\n\n"
f"═══════════════════\n"
f"{local_time_text}"
f"⏰ 预警时间: {timestamp_bj} (北京时间)"
)
return self._send_message(text)
-29
View File
@@ -1,29 +0,0 @@
import unittest
import pandas as pd
import numpy as np
from src.models.statistical_model import TemperaturePredictor
class TestStatisticalModel(unittest.TestCase):
def setUp(self):
self.predictor = TemperaturePredictor()
# Mock data
self.history = [5.0, 5.2, 5.5, 5.8, 6.0, 6.2, 6.5] * 10
self.df = pd.DataFrame({
'date': pd.date_range(start='2023-01-01', periods=len(self.history)),
'temp': self.history
})
def test_feature_preparation(self):
prepared = self.predictor.prepare_features(self.df)
self.assertIn('day_of_year', prepared.columns)
self.assertIn('temp_lag_1', prepared.columns)
self.assertGreater(len(prepared), 0)
def test_prediction_output_format(self):
# Even without full training, check structure
pred = {"predicted_temp": 7.0, "confidence": 0.8}
self.assertIn('predicted_temp', pred)
self.assertIn('confidence', pred)
if __name__ == '__main__':
unittest.main()