diff --git a/api/__init__.py b/api/__init__.py new file mode 100644 index 0000000..d25ee0b --- /dev/null +++ b/api/__init__.py @@ -0,0 +1 @@ +# api模块初始化 diff --git a/api/polymarket_client.py b/api/polymarket_client.py new file mode 100644 index 0000000..185341f --- /dev/null +++ b/api/polymarket_client.py @@ -0,0 +1,390 @@ +""" +Polymarket API 客户端 + +用于与Polymarket CLOB API交互，支持： +- 搜索BTC相关市场 +- 获取订单簿和价格 +- 下单交易 +""" + +import os +import httpx +from typing import Optional, Dict, List, Any +from dataclasses import dataclass +from dotenv import load_dotenv + +# 加载环境变量 +load_dotenv() + + +@dataclass +class MarketInfo: + """市场信息""" + condition_id: str + question: str + outcomes: List[str] + outcome_prices: List[float] + tokens: List[Dict] + volume: float + liquidity: float + end_date: str + + +class PolymarketClient: + """Polymarket API客户端""" + + GAMMA_API = "https://gamma-api.polymarket.com" + CLOB_API = "https://clob.polymarket.com" + + def __init__(self, private_key: Optional[str] = None): + """ + 初始化客户端 + + Args: + private_key: 钱包私钥(用于交易)，如不提供则只能读取数据 + """ + self.private_key = private_key or os.getenv("POLYMARKET_PRIVATE_KEY") + + # 配置代理 + proxy = os.getenv("SOCKS5_PROXY") if os.getenv("USE_PROXY", "false").lower() == "true" else None + + # HTTP客户端 - 禁用压缩避免brotli问题 + self.http = httpx.Client( + timeout=30.0, + proxy=proxy, + headers={'Accept-Encoding': 'identity'} # 禁用压缩 + ) + + # 如果有私钥，初始化CLOB客户端 + self.clob_client = None + if self.private_key: + self._init_clob_client() + + def _init_clob_client(self): + """初始化CLOB交易客户端""" + try: + from py_clob_client.client import ClobClient + from py_clob_client.clob_types import ApiCreds + + # Polygon链ID + chain_id = 137 + + # 初始化客户端 + self.clob_client = ClobClient( + host=self.CLOB_API, + key=self.private_key, + chain_id=chain_id, + ) + + # 获取或创建API凭证 + self.clob_client.set_api_creds(self.clob_client.create_or_derive_api_creds()) + + print("✓ CLOB客户端初始化成功") + except Exception as e: + print(f"⚠️ CLOB客户端初始化失败: {e}") + self.clob_client = None + + def search_markets(self, query: str = "BTC", limit: int = 20) -> List[Dict]: + """ + 搜索市场 + + Args: + query: 搜索关键词 + limit: 返回数量限制 + + Returns: + 市场列表 + """ + try: + response = self.http.get( + f"{self.GAMMA_API}/markets", + params={ + "active": "true", + "closed": "false", + "limit": limit, + } + ) + response.raise_for_status() + markets = response.json() + + # 过滤包含关键词的市场 + filtered = [m for m in markets if query.lower() in m.get("question", "").lower()] + return filtered + except Exception as e: + print(f"搜索市场失败: {e}") + return [] + + def search_btc_1h_markets(self) -> List[Dict]: + """ + 搜索BTC 1小时价格预测市场 + + Returns: + BTC 1H相关市场列表 + """ + try: + response = self.http.get( + f"{self.GAMMA_API}/markets", + params={ + "active": "true", + "closed": "false", + "limit": 100, + } + ) + response.raise_for_status() + markets = response.json() + + # 过滤BTC 1小时相关市场 + btc_1h_markets = [] + keywords = ["btc", "bitcoin"] + time_keywords = ["1h", "1 hour", "hourly", "hour"] + + for m in markets: + question = m.get("question", "").lower() + # 检查是否包含BTC关键词 + has_btc = any(kw in question for kw in keywords) + # 检查是否包含1小时关键词 + has_1h = any(kw in question for kw in time_keywords) + + if has_btc and has_1h: + btc_1h_markets.append(m) + + return btc_1h_markets + except Exception as e: + print(f"搜索BTC 1H市场失败: {e}") + return [] + + def get_market_info(self, condition_id: str) -> Optional[MarketInfo]: + """ + 获取市场详细信息 + + Args: + condition_id: 市场条件ID + + Returns: + 市场信息对象 + """ + try: + response = self.http.get(f"{self.GAMMA_API}/markets/{condition_id}") + response.raise_for_status() + data = response.json() + + return MarketInfo( + condition_id=data.get("conditionId", ""), + question=data.get("question", ""), + outcomes=data.get("outcomes", []), + outcome_prices=data.get("outcomePrices", []), + tokens=data.get("tokens", []), + volume=float(data.get("volume", 0)), + liquidity=float(data.get("liquidity", 0)), + end_date=data.get("endDate", ""), + ) + except Exception as e: + print(f"获取市场信息失败: {e}") + return None + + def get_orderbook(self, token_id: str) -> Dict: + """ + 获取订单簿 + + Args: + token_id: 代币ID + + Returns: + 订单簿数据 + """ + try: + response = self.http.get( + f"{self.CLOB_API}/book", + params={"token_id": token_id} + ) + response.raise_for_status() + return response.json() + except Exception as e: + print(f"获取订单簿失败: {e}") + return {} + + def get_price(self, token_id: str) -> Optional[float]: + """ + 获取代币当前价格 + + Args: + token_id: 代币ID + + Returns: + 当前价格(0-1之间) + """ + try: + response = self.http.get( + f"{self.CLOB_API}/price", + params={"token_id": token_id, "side": "buy"} + ) + response.raise_for_status() + data = response.json() + return float(data.get("price", 0)) + except Exception as e: + print(f"获取价格失败: {e}") + return None + + def place_market_order( + self, + token_id: str, + side: str, # "BUY" or "SELL" + amount: float, # USDC金额 + ) -> Optional[Dict]: + """ + 下市价单 + + Args: + token_id: 代币ID + side: 买卖方向 "BUY" 或 "SELL" + amount: 下注金额(USDC) + + Returns: + 订单响应 + """ + if not self.clob_client: + print("❌ CLOB客户端未初始化，无法下单") + return None + + try: + from py_clob_client.order_builder.constants import BUY, SELL + + order_side = BUY if side.upper() == "BUY" else SELL + + # 创建市价单 + order = self.clob_client.create_market_order( + token_id=token_id, + amount=amount, + side=order_side, + ) + + # 提交订单 + response = self.clob_client.post_order(order) + print(f"✓ 订单已提交: {response}") + return response + except Exception as e: + print(f"❌ 下单失败: {e}") + return None + + def place_limit_order( + self, + token_id: str, + side: str, # "BUY" or "SELL" + price: float, # 价格(0-1) + size: float, # 数量 + ) -> Optional[Dict]: + """ + 下限价单 + + Args: + token_id: 代币ID + side: 买卖方向 "BUY" 或 "SELL" + price: 价格(0-1之间) + size: 购买数量 + + Returns: + 订单响应 + """ + if not self.clob_client: + print("❌ CLOB客户端未初始化，无法下单") + return None + + try: + from py_clob_client.order_builder.constants import BUY, SELL + + order_side = BUY if side.upper() == "BUY" else SELL + + # 创建限价单 + order = self.clob_client.create_limit_order( + token_id=token_id, + price=price, + size=size, + side=order_side, + ) + + # 提交订单 + response = self.clob_client.post_order(order) + print(f"✓ 限价单已提交: {response}") + return response + except Exception as e: + print(f"❌ 下单失败: {e}") + return None + + def get_positions(self) -> List[Dict]: + """ + 获取当前仓位 + + Returns: + 仓位列表 + """ + if not self.clob_client: + print("❌ CLOB客户端未初始化") + return [] + + try: + positions = self.clob_client.get_positions() + return positions + except Exception as e: + print(f"获取仓位失败: {e}") + return [] + + def cancel_order(self, order_id: str) -> bool: + """ + 取消订单 + + Args: + order_id: 订单ID + + Returns: + 是否成功 + """ + if not self.clob_client: + print("❌ CLOB客户端未初始化") + return False + + try: + self.clob_client.cancel(order_id) + print(f"✓ 订单已取消: {order_id}") + return True + except Exception as e: + print(f"❌ 取消订单失败: {e}") + return False + + def close(self): + """关闭客户端""" + self.http.close() + + +# 测试函数 +def test_search_markets(): + """测试搜索市场功能""" + client = PolymarketClient() + + print("搜索BTC相关市场...") + markets = client.search_markets("BTC", limit=10) + + if markets: + print(f"\n找到 {len(markets)} 个BTC相关市场:\n") + for m in markets[:5]: + print(f" 问题: {m.get('question', 'N/A')}") + print(f" 条件ID: {m.get('conditionId', 'N/A')}") + print(f" 结果: {m.get('outcomes', [])}") + print(f" 价格: {m.get('outcomePrices', [])}") + print("-" * 50) + else: + print("未找到BTC相关市场") + + print("\n搜索BTC 1H市场...") + btc_1h = client.search_btc_1h_markets() + if btc_1h: + print(f"找到 {len(btc_1h)} 个BTC 1H市场") + for m in btc_1h: + print(f" {m.get('question', 'N/A')}") + else: + print("未找到BTC 1H市场") + + client.close() + + +if __name__ == "__main__": + test_search_markets() diff --git a/auto_trade_polymarket.py b/auto_trade_polymarket.py new file mode 100644 index 0000000..48b8a32 --- /dev/null +++ b/auto_trade_polymarket.py @@ -0,0 +1,844 @@ +""" +BTC Hourly 自动交易脚本 + +基于DLinear模型预测，在Polymarket自动下单BTC小时涨跌 +支持: +- 每小时自动寻找对应市场 +- 实时价格刷新 +- 定时下单 +- 详细日志记录 +""" + +import os +import sys +import argparse +import json +import time +import logging +from logging.handlers import RotatingFileHandler +import schedule +from datetime import datetime, timedelta +from typing import Optional, Tuple, Dict +import pickle +from zoneinfo import ZoneInfo + +import torch +import numpy as np +import pandas as pd +import pandas_ta as ta +from sklearn.preprocessing import StandardScaler +from dotenv import load_dotenv +import httpx + +# 添加项目路径 +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from data.fetcher import BinanceDataFetcher + +load_dotenv() + + +def setup_logging(log_dir: str = "logs", log_level: int = logging.INFO) -> logging.Logger: + """ + 配置日志系统 + + Args: + log_dir: 日志目录 + log_level: 日志级别 + + Returns: + 配置好的logger + """ + # 创建日志目录 + os.makedirs(log_dir, exist_ok=True) + + # 创建logger + logger = logging.getLogger("polymarket_trader") + logger.setLevel(log_level) + + # 避免重复添加handler + if logger.handlers: + return logger + + # 日志格式 + formatter = logging.Formatter( + "%(asctime)s | %(levelname)-8s | %(message)s", + datefmt="%Y-%m-%d %H:%M:%S" + ) + + # 文件handler (按大小轮转, 最大10MB, 保留5个备份) + log_file = os.path.join(log_dir, "polymarket_trading.log") + file_handler = RotatingFileHandler( + log_file, + maxBytes=10*1024*1024, # 10MB + backupCount=5, + encoding='utf-8' + ) + file_handler.setLevel(log_level) + file_handler.setFormatter(formatter) + logger.addHandler(file_handler) + + # 控制台handler + console_handler = logging.StreamHandler(sys.stdout) + console_handler.setLevel(log_level) + console_handler.setFormatter(formatter) + logger.addHandler(console_handler) + + return logger + + +# 初始化全局logger +logger = setup_logging() + + + +class ClassificationPreprocessor: + """DLinear分类预处理器""" + + def __init__(self, lookback: int = 168, forecast_horizon: int = 1): + self.lookback = lookback + self.forecast_horizon = forecast_horizon + self.scaler = StandardScaler() + self.is_fitted = False + + def add_features(self, df: pd.DataFrame) -> pd.DataFrame: + df = df.copy() + df['rsi'] = ta.rsi(df['close'], length=14) + macd = ta.macd(df['close']) + if macd is not None and not macd.empty: + df['macd'] = macd.iloc[:, 0] + df['macd_hist'] = macd.iloc[:, 1] + df['macd_signal'] = macd.iloc[:, 2] + return df + + def create_labels(self, df: pd.DataFrame) -> pd.DataFrame: + df = df.copy() + df['future_price'] = df['close'].shift(-self.forecast_horizon) + df['future_return'] = (df['future_price'] - df['close']) / df['close'] + df['label'] = (df['future_return'] > 0).astype(int) + return df[:-self.forecast_horizon] + + def preprocess(self, df: pd.DataFrame, fit: bool = True): + df = self.add_features(df) + df = self.create_labels(df) + df = df.dropna().reset_index(drop=True) + + feature_cols = ['open', 'high', 'low', 'close', 'volume', + 'rsi', 'macd', 'macd_hist', 'macd_signal'] + + if fit: + scaled_features = self.scaler.fit_transform(df[feature_cols]) + self.is_fitted = True + self.feature_columns = feature_cols + else: + scaled_features = self.scaler.transform(df[feature_cols]) + + X, y = [], [] + for i in range(len(scaled_features) - self.lookback): + X.append(scaled_features[i:i + self.lookback]) + y.append(df['label'].iloc[i + self.lookback]) + + return np.array(X), np.array(y), df + + +class PolymarketBTCTrader: + """Polymarket BTC Hourly交易器""" + + GAMMA_API = "https://gamma-api.polymarket.com" + CLOB_API = "https://clob.polymarket.com" + + # 月份名称映射 + MONTH_NAMES = { + 1: 'january', 2: 'february', 3: 'march', 4: 'april', + 5: 'may', 6: 'june', 7: 'july', 8: 'august', + 9: 'september', 10: 'october', 11: 'november', 12: 'december' + } + + def __init__( + self, + private_key: Optional[str] = None, + max_bet: float = 1.0, + max_price: float = 0.50, + dry_run: bool = True, + ): + self.private_key = private_key or os.getenv("POLYMARKET_PRIVATE_KEY") + self.max_bet = max_bet + self.max_price = max_price + self.dry_run = dry_run + + # HTTP客户端 + proxy = os.getenv("SOCKS5_PROXY") if os.getenv("USE_PROXY", "false").lower() == "true" else None + self.http = httpx.Client( + timeout=30.0, + proxy=proxy, + headers={'Accept-Encoding': 'identity'} + ) + + # 设备 + self.device = torch.device('mps' if torch.backends.mps.is_available() else + 'cuda' if torch.cuda.is_available() else 'cpu') + + # 加载模型 + self.model = None + self.preprocessor = None + self._load_model() + + # 数据获取器 + self.fetcher = BinanceDataFetcher() + + # CLOB客户端(用于下单) + self.clob_client = None + if self.private_key and not self.dry_run: + self._init_clob_client() + + # 当前市场缓存 + self.current_market = None + self.last_price_update = None + + def _load_model(self): + """加载DLinear模型""" + logger.info("加载DLinear模型...") + + from models.dlinear.model import DLinear + + checkpoint_path = './models/checkpoints/dlinear_classification_2026/best_model.pt' + preprocessor_path = './models/checkpoints/dlinear_classification_2026/preprocessor.pkl' + + checkpoint = torch.load(checkpoint_path, map_location=self.device, weights_only=False) + self.model = DLinear(checkpoint['config']) + self.model.load_state_dict(checkpoint['model_state_dict']) + self.model = self.model.to(self.device).eval() + + with open(preprocessor_path, 'rb') as f: + self.preprocessor = pickle.load(f) + + logger.info(f"✓ 模型已加载 (设备: {self.device})") + + def _init_clob_client(self): + """初始化CLOB交易客户端""" + try: + from py_clob_client.client import ClobClient + + self.clob_client = ClobClient( + host=self.CLOB_API, + key=self.private_key, + chain_id=137, # Polygon + ) + self.clob_client.set_api_creds(self.clob_client.create_or_derive_api_creds()) + logger.info("✓ CLOB客户端初始化成功") + except Exception as e: + logger.warning(f"⚠️ CLOB客户端初始化失败: {e}") + self.clob_client = None + + def generate_market_slug(self, target_time: Optional[datetime] = None) -> str: + """ + 生成市场slug + + 格式: bitcoin-up-or-down-{month}-{day}-{hour}am/pm-et + 例如: bitcoin-up-or-down-february-3-8am-et + + Args: + target_time: 目标时间(ET时区) + for_current_hour: 如果True，使用当前整点；否则使用下一个整点 + + Returns: + 市场slug + """ + # 获取美东时间 + et_tz = ZoneInfo('America/New_York') + + if target_time is None: + now_et = datetime.now(et_tz) + target_time = now_et.replace(minute=0, second=0, microsecond=0) + + month_name = self.MONTH_NAMES[target_time.month] + day = target_time.day + + # 转换为12小时制 + hour_24 = target_time.hour + if hour_24 == 0: + hour_12 = 12 + am_pm = 'am' + elif hour_24 < 12: + hour_12 = hour_24 + am_pm = 'am' + elif hour_24 == 12: + hour_12 = 12 + am_pm = 'pm' + else: + hour_12 = hour_24 - 12 + am_pm = 'pm' + + slug = f"bitcoin-up-or-down-{month_name}-{day}-{hour_12}{am_pm}-et" + return slug + + def get_market_by_slug(self, slug: str) -> Optional[Dict]: + """ + 通过slug获取市场信息 + + Args: + slug: 市场slug + + Returns: + 市场信息 + """ + try: + response = self.http.get( + f"{self.GAMMA_API}/events", + params={'slug': slug} + ) + response.raise_for_status() + events = response.json() + + if not events: + return None + + event = events[0] + markets = event.get('markets', []) + + if not markets: + return None + + market = markets[0] + + # outcomePrices可能是JSON字符串或列表 + outcome_prices_raw = market.get('outcomePrices', ['0', '0']) + if isinstance(outcome_prices_raw, str): + import json as _json + try: + outcome_prices_raw = _json.loads(outcome_prices_raw) + except: + outcome_prices_raw = ['0', '0'] + + # clobTokenIds也可能是JSON字符串 + clob_tokens_raw = market.get('clobTokenIds', []) + if isinstance(clob_tokens_raw, str): + import json as _json + try: + clob_tokens_raw = _json.loads(clob_tokens_raw) + except: + clob_tokens_raw = [] + + return { + 'title': event.get('title', ''), + 'slug': slug, + 'question': market.get('question', ''), + 'condition_id': market.get('conditionId', ''), + 'outcomes': market.get('outcomes', []), + 'outcome_prices': [float(p) for p in outcome_prices_raw], + 'tokens': clob_tokens_raw, + 'end_date': event.get('endDate', ''), + 'closed': market.get('closed', False), + } + except Exception as e: + logger.error(f"获取市场失败: {e}") + return None + + def get_current_hourly_market(self) -> Optional[Dict]: + """ + 获取BTC市场 + + Args: + for_current_hour: 如果True，获取当前整点市场；否则获取下一个整点市场 + + Returns: + 市场信息 + """ + slug = self.generate_market_slug() + logger.info(f"搜索市场: {slug}") + + market = self.get_market_by_slug(slug) + + if market: + self.current_market = market + self.last_price_update = datetime.now() + logger.info(f"✓ 找到市场: {market['title']}") + logger.info(f" Up价格: ${market['outcome_prices'][0]:.3f}") + logger.info(f" Down价格: ${market['outcome_prices'][1]:.3f}") + else: + logger.warning(f"❌ 未找到市场: {slug}") + + return market + + def refresh_prices(self) -> Optional[Dict]: + """ + 刷新当前市场价格 + + Returns: + 更新后的市场信息 + """ + if not self.current_market: + self.get_current_hourly_market() + + # 使用CLOB API获取实时价格 + tokens = self.current_market.get('tokens', []) + if len(tokens) >= 2: + realtime_prices = self.get_realtime_prices(tokens[0], tokens[1]) + if realtime_prices: + old_prices = self.current_market['outcome_prices'] + new_prices = realtime_prices + + self.current_market['outcome_prices'] = new_prices + self.last_price_update = datetime.now() + + # 显示价格变化 + up_change = new_prices[0] - old_prices[0] + down_change = new_prices[1] - old_prices[1] + + logger.info(f"📊 实时价格 [{datetime.now().strftime('%H:%M:%S')}]") + logger.info(f" Up: ${new_prices[0]:.3f} ({up_change:+.3f})") + logger.info(f" Down: ${new_prices[1]:.3f} ({down_change:+.3f})") + + return self.current_market + + # 备用: 使用Gamma API + slug = self.current_market['slug'] + market = self.get_market_by_slug(slug) + + if market: + old_prices = self.current_market['outcome_prices'] + new_prices = market['outcome_prices'] + + self.current_market = market + self.last_price_update = datetime.now() + + # 显示价格变化 + up_change = new_prices[0] - old_prices[0] + down_change = new_prices[1] - old_prices[1] + + logger.info(f"📊 价格更新(Gamma) [{datetime.now().strftime('%H:%M:%S')}]") + logger.info(f" Up: ${new_prices[0]:.3f} ({up_change:+.3f})") + logger.info(f" Down: ${new_prices[1]:.3f} ({down_change:+.3f})") + + return market + + def get_realtime_prices(self, up_token_id: str, down_token_id: str) -> Optional[list]: + """ + 从CLOB API获取实时买入价格 + + 使用/price?side=BUY端点获取真实买入价格 (与页面显示一致) + + Args: + up_token_id: Up代币ID + down_token_id: Down代币ID + + Returns: + [up_price, down_price] 或 None + """ + try: + # 使用/price端点获取买入价格 + up_response = self.http.get( + f"{self.CLOB_API}/price", + params={'token_id': up_token_id, 'side': 'BUY'} + ) + up_response.raise_for_status() + up_data = up_response.json() + + down_response = self.http.get( + f"{self.CLOB_API}/price", + params={'token_id': down_token_id, 'side': 'BUY'} + ) + down_response.raise_for_status() + down_data = down_response.json() + + # 解析买入价格 + up_price = float(up_data.get('price', 0.5)) + down_price = float(down_data.get('price', 0.5)) + + return [up_price, down_price] + except Exception as e: + logger.debug(f"获取实时价格失败: {e}") + return None + + def execute_trade(self, prediction: int, market: Dict) -> Dict: + """ + 执行交易 + + Args: + prediction: 预测结果(0=跌, 1=涨) + market: 市场信息 + + Returns: + 交易结果 + """ + result = { + 'timestamp': datetime.now().isoformat(), + 'market': market['title'], + 'prediction': 'UP' if prediction == 1 else 'DOWN', + 'success': False, + 'order': None, + 'error': None, + } + + if prediction == 1: # 预测涨 -> 买UP + token_id = market['tokens'][0] + price = market['outcome_prices'][0] + side = "UP" + else: # 预测跌 -> 买DOWN + token_id = market['tokens'][1] + price = market['outcome_prices'][1] + side = "DOWN" + + result['side'] = side + result['price'] = price + + # 检查价格限制 + if price > self.max_price: + result['error'] = f"价格${price:.2f}超过限制${self.max_price:.2f}" + logger.error(f"❌ {result['error']}") + return result + + # 计算数量 + size = self.max_bet / price + result['size'] = size + result['amount'] = self.max_bet + + logger.info("📋 交易详情:") + logger.info(f" 方向: {side}") + logger.info(f" 价格: ${price:.3f}") + logger.info(f" 金额: ${self.max_bet:.2f} USDC") + logger.info(f" 数量: {size:.2f} 股") + + if self.dry_run: + logger.info("🔸 [模拟模式] 未实际下单") + result['success'] = True + result['order'] = 'DRY_RUN' + return result + + # 实际下单 + if not self.clob_client: + result['error'] = "CLOB客户端未初始化" + logger.error(f"❌ {result['error']}") + return result + + try: + from py_clob_client.order_builder.constants import BUY + + order = self.clob_client.create_limit_order( + token_id=token_id, + price=price, + size=size, + side=BUY, + ) + + response = self.clob_client.post_order(order) + logger.info(f"✓ 订单已提交: {response}") + result['success'] = True + result['order'] = response + except Exception as e: + result['error'] = str(e) + logger.error(f"❌ 下单失败: {e}") + + return result + + def get_prediction_with_delay(self, delay_seconds: float = 2.0) -> Tuple[int, float]: + """ + 延迟后获取K线并预测当前小时的涨跌 + + 注意: 使用上一小时收盘的K线(倒数第二根)来预测当前小时 + 例如: 22:00整点时，用21:00收盘的K线预测22:00这个小时的涨跌 + + Args: + delay_seconds: 延迟秒数(等待K线更新) + + Returns: + (prediction, probability): 预测结果(0=跌, 1=涨)和概率 + """ + logger.info(f"⏳ 等待{delay_seconds}秒获取最新K线...") + time.sleep(delay_seconds) + + logger.info("获取最新BTC数据...") + + end_date = (datetime.now() + timedelta(days=1)).strftime('%Y-%m-%d') + start_date = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d') + + df = self.fetcher.fetch_ohlcv( + symbol='BTC', + timeframe='1h', + start_date=start_date, + end_date=end_date, + use_cache=False # 不使用缓存，获取最新数据 + ) + + # 使用倒数第二根K线(上一小时收盘)进行预测 + # 例如22:00时，最新K线是14:00 UTC(22:00北京)，我们用13:00 UTC(21:00北京)的数据 + prev_time = df['timestamp'].iloc[-2] + prev_close = df['close'].iloc[-2] + current_time = df['timestamp'].iloc[-1] + current_close = df['close'].iloc[-1] + + logger.info(f" 上一K线时间: {prev_time} (用于预测)") + logger.info(f" 上一收盘价: ${prev_close:,.2f}") + logger.info(f" 当前K线时间: {current_time} (预测目标)") + logger.info(f" 当前收盘价: ${current_close:,.2f} (实际结果)") + + # 用除最后一根外的数据进行预测 + df_for_prediction = df.iloc[:-1] + X, _, _ = self.preprocessor.preprocess(df_for_prediction, fit=False) + X_last = torch.FloatTensor(X[-1:]).to(self.device) + + with torch.no_grad(): + logits = self.model(X_last).cpu().numpy() + prob = 1 / (1 + np.exp(-logits)) + prediction = int(prob.item() > 0.5) + probability = float(prob.item()) + + direction = "📈 UP (涨)" if prediction == 1 else "📉 DOWN (跌)" + confidence = probability if prediction == 1 else 1 - probability + logger.info(f"🤖 DLinear预测: {direction}") + logger.info(f" 置信度: {confidence:.2%}") + + # 显示预测是否正确 (用于验证) + actual_up = current_close > prev_close + if prediction == 1: + correct = actual_up + else: + correct = not actual_up + logger.info(f" 实际涨跌: {'📈 涨' if actual_up else '📉 跌'} ({'✅ 正确' if correct else '❌ 错误'})") + + return prediction, probability + + def run_at_hour_start(self, delay_seconds: float = 2.0) -> Dict: + """ + 整点时执行的交易流程 + + 1. 延迟delay_seconds秒等待K线更新 + 2. 获取最新K线 + 3. DLinear预测当前整点的收盘涨跌 + 4. 获取对应的Polymarket市场 + 5. 检查价格是否≤max_price + 6. 下单 + + Args: + delay_seconds: 整点后延迟秒数 + + Returns: + 执行结果 + """ + logger.info("=" * 60) + logger.info(f"⏰ 整点交易触发 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") + logger.info("=" * 60) + logger.info(f" 模式: {'🔸 模拟' if self.dry_run else '💰 真实交易'}") + logger.info(f" 最大下注: ${self.max_bet:.2f} USDC") + logger.info(f" 最高价格: ${self.max_price:.2f}") + + result = { + 'timestamp': datetime.now().isoformat(), + 'success': False, + 'error': None, + } + + # 1. 延迟后获取预测 + prediction, probability = self.get_prediction_with_delay(delay_seconds) + result['prediction'] = 'UP' if prediction == 1 else 'DOWN' + result['probability'] = probability if prediction == 1 else 1 - probability + + # 2. 获取当前整点的市场(不是下一个整点) + market = self.get_current_hourly_market() + if not market: + result['error'] = '未找到市场' + logger.error(f"❌ {result['error']}") + return result + + result['market'] = market['title'] + + if market['closed']: + result['error'] = '市场已关闭' + logger.error(f"❌ {result['error']}") + return result + + # 3. 执行交易 + trade_result = self.execute_trade(prediction, market) + result.update(trade_result) + + logger.info("=" * 60) + + return result + + def _show_initial_prediction(self): + """ + 启动时显示当前预测（不下单） + """ + logger.info("\n📊 当前预测状态:") + + end_date = (datetime.now() + timedelta(days=1)).strftime('%Y-%m-%d') + start_date = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d') + + df = self.fetcher.fetch_ohlcv( + symbol='BTC', + timeframe='1h', + start_date=start_date, + end_date=end_date, + use_cache=False + ) + + # 显示K线信息 + prev_time = df['timestamp'].iloc[-2] + prev_close = df['close'].iloc[-2] + current_time = df['timestamp'].iloc[-1] + current_close = df['close'].iloc[-1] + + logger.info(f" 上一K线: {prev_time} 收盘价: ${prev_close:,.2f}") + logger.info(f" 当前K线: {current_time} 收盘价: ${current_close:,.2f}") + + # 进行预测 + df_for_prediction = df.iloc[:-1] + X, _, _ = self.preprocessor.preprocess(df_for_prediction, fit=False) + X_last = torch.FloatTensor(X[-1:]).to(self.device) + + with torch.no_grad(): + logits = self.model(X_last).cpu().numpy() + prob = 1 / (1 + np.exp(-logits)) + prediction = int(prob.item() > 0.5) + probability = float(prob.item()) + + direction = "📈 UP (涨)" if prediction == 1 else "📉 DOWN (跌)" + confidence = probability if prediction == 1 else 1 - probability + + logger.info(f" 🤖 预测: {direction} (置信度: {confidence:.2%})") + + # 显示市场价格 + if self.current_market: + up_price = self.current_market['outcome_prices'][0] + down_price = self.current_market['outcome_prices'][1] + if prediction == 1: + logger.info(f" 💰 买入Up价格: ${up_price:.3f} {'✅ 可下单' if up_price <= self.max_price else '❌ 超过限价'}") + else: + logger.info(f" 💰 买入Down价格: ${down_price:.3f} {'✅ 可下单' if down_price <= self.max_price else '❌ 超过限价'}") + + # 计算距下一整点时间 + now = datetime.now() + next_hour = now.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1) + wait_minutes = int((next_hour - now).total_seconds() / 60) + logger.info(f" ⏰ 距离下一整点: {wait_minutes}分钟") + logger.info("") + + def wait_for_next_hour(self) -> datetime: + """ + 等待到下一个整点 + + Returns: + 下一个整点时间 + """ + now = datetime.now() + next_hour = now.replace(minute=0, second=0, microsecond=0) + timedelta(hours=1) + wait_seconds = (next_hour - now).total_seconds() + + logger.info(f"⏳ 等待下一个整点: {next_hour.strftime('%H:%M:%S')}") + logger.info(f" 剩余等待: {int(wait_seconds)}秒") + + return next_hour + + def run_scheduler(self, interval_minutes: int = 5, delay_seconds: float = 2.0): + """ + 定时运行交易器 + + 工作流程: + 1. 每个整点+delay_seconds秒触发交易 + 2. 期间每interval_minutes分钟刷新价格 + + Args: + interval_minutes: 价格刷新间隔(分钟) + delay_seconds: 整点后延迟秒数 + """ + logger.info("=" * 60) + logger.info("🤖 BTC Hourly自动交易已启动") + logger.info("=" * 60) + logger.info(f" 模式: {'🔸 模拟' if self.dry_run else '💰 真实交易'}") + logger.info(f" 下单时机: 每个整点后延迟{delay_seconds}秒") + logger.info(f" 价格刷新: 每{interval_minutes}分钟") + logger.info(f" 最大下注: ${self.max_bet:.2f} USDC") + logger.info(f" 最高价格: ${self.max_price:.2f}") + logger.info(" 按 Ctrl+C 停止") + logger.info("=" * 60) + + # 立即获取整点的市场 + self.get_current_hourly_market() + + # 启动时进行一次预测（不下单，仅显示） + self._show_initial_prediction() + + # 定时刷新价格 + schedule.every(interval_minutes).minutes.do(self.refresh_prices) + + # 每小时整点后执行交易 + def hourly_trade(): + self.run_at_hour_start(delay_seconds=delay_seconds) + + schedule.every().hour.at(":00").do(lambda: time.sleep(delay_seconds) or hourly_trade()) + + try: + while True: + schedule.run_pending() + time.sleep(1) + except KeyboardInterrupt: + logger.info("✓ 交易器已停止") + + def close(self): + """关闭资源""" + self.http.close() + + +def main(): + parser = argparse.ArgumentParser(description='BTC Hourly自动交易 - 基于DLinear模型') + parser.add_argument('--dry-run', action='store_true', default=True, + help='模拟模式(不实际下单)') + parser.add_argument('--real', action='store_true', + help='真实交易模式') + parser.add_argument('--max-bet', type=float, default=1.0, + help='最大下注金额(USDC)') + parser.add_argument('--max-price', type=float, default=0.50, + help='最高可接受价格') + parser.add_argument('--once', action='store_true', + help='只执行一次(不启动定时器)') + parser.add_argument('--interval', type=int, default=1, + help='价格刷新间隔(分钟)') + parser.add_argument('--check-market', action='store_true', + help='只检查当前市场信息') + + args = parser.parse_args() + + dry_run = not args.real + + # 真实交易确认 + if not dry_run: + logger.warning("⚠️ 警告: 即将进行真实交易!") + logger.warning(f" 最大下注: ${args.max_bet:.2f} USDC") + confirm = input("确认继续? (yes/no): ") + if confirm.lower() != 'yes': + logger.info("已取消") + return + + trader = PolymarketBTCTrader( + max_bet=args.max_bet, + max_price=args.max_price, + dry_run=dry_run, + ) + + try: + if args.check_market: + # 只检查市场 + trader.get_current_hourly_market() + elif args.once: + # 执行一次 + result = trader.run_once() + + # 保存日志 + log_file = 'trading_log.json' + logs = [] + if os.path.exists(log_file): + with open(log_file, 'r') as f: + logs = json.load(f) + logs.append(result) + with open(log_file, 'w') as f: + json.dump(logs, f, indent=2, ensure_ascii=False) + logger.info(f"结果已保存到 {log_file}") + else: + # 定时运行 + trader.run_scheduler(interval_minutes=args.interval) + finally: + trader.close() + + +if __name__ == "__main__": + main() diff --git a/data/database.py b/data/database.py new file mode 100644 index 0000000..88113a6 --- /dev/null +++ b/data/database.py @@ -0,0 +1,139 @@ +import sqlite3 +import pandas as pd +from pathlib import Path +from typing import Optional, List, Tuple +from contextlib import contextmanager +import os + +class DatabaseManager: + """Manages SQLite database for storing OHLCV data.""" + + def __init__(self, db_path: str = "./data/cache/market_data.db"): + self.db_path = Path(db_path) + self.db_path.parent.mkdir(parents=True, exist_ok=True) + self.initialize_db() + + @contextmanager + def get_connection(self): + """Context manager for database connection.""" + conn = sqlite3.connect(self.db_path) + try: + yield conn + finally: + conn.close() + + def initialize_db(self): + """Create tables if they don't exist.""" + with self.get_connection() as conn: + cursor = conn.cursor() + + # Create OHLCV table + # Composite primary key ensures uniqueness for a specific candle + cursor.execute(""" + CREATE TABLE IF NOT EXISTS ohlcv ( + timestamp DATETIME, + symbol TEXT, + timeframe TEXT, + open REAL, + high REAL, + low REAL, + close REAL, + volume REAL, + PRIMARY KEY (symbol, timeframe, timestamp) + ) + """) + + # Create index for faster querying by time range + cursor.execute(""" + CREATE INDEX IF NOT EXISTS idx_ohlcv_time + ON ohlcv (symbol, timeframe, timestamp) + """) + + conn.commit() + + def save_ohlcv(self, df: pd.DataFrame, symbol: str, timeframe: str): + """Save OHLCV data to database (Upsert).""" + if df.empty: + return + + # Ensure timestamp is datetime + df = df.copy() + if not pd.api.types.is_datetime64_any_dtype(df['timestamp']): + df['timestamp'] = pd.to_datetime(df['timestamp']) + + # Convert to string for SQLite compatibility (ISO 8601) + df['timestamp'] = df['timestamp'].dt.strftime('%Y-%m-%d %H:%M:%S') + + # Add metadata columns if they don't exist + if 'symbol' not in df.columns: + df['symbol'] = symbol + if 'timeframe' not in df.columns: + df['timeframe'] = timeframe + + # Prepare data for insertion (ensure order) + records = df[['timestamp', 'symbol', 'timeframe', 'open', 'high', 'low', 'close', 'volume']].to_records(index=False) + + with self.get_connection() as conn: + cursor = conn.cursor() + # efficiently insert or replace + # sqlite3 adapter handles datetime objects correctly + cursor.executemany(""" + INSERT OR REPLACE INTO ohlcv + (timestamp, symbol, timeframe, open, high, low, close, volume) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, records) + conn.commit() + + def get_ohlcv(self, symbol: str, timeframe: str, start_date: Optional[str] = None, end_date: Optional[str] = None) -> pd.DataFrame: + """Retrieve OHLCV data from database.""" + query = """ + SELECT timestamp, open, high, low, close, volume + FROM ohlcv + WHERE symbol = ? AND timeframe = ? + """ + params = [symbol, timeframe] + + if start_date: + query += " AND timestamp >= ?" + params.append(start_date) + + if end_date: + # Handle 'YYYY-MM-DD' vs full timestamp + if len(str(end_date)) == 10: # YYYY-MM-DD + query += " AND timestamp < date(?, '+1 day')" + else: + query += " AND timestamp <= ?" + params.append(end_date) + + query += " ORDER BY timestamp ASC" + + with self.get_connection() as conn: + df = pd.read_sql_query(query, conn, params=params, parse_dates=['timestamp']) + + return df + + def get_latest_timestamp(self, symbol: str, timeframe: str) -> Optional[pd.Timestamp]: + """Get the latest timestamp for a symbol/timeframe.""" + with self.get_connection() as conn: + cursor = conn.cursor() + cursor.execute(""" + SELECT MAX(timestamp) + FROM ohlcv + WHERE symbol = ? AND timeframe = ? + """, (symbol, timeframe)) + result = cursor.fetchone() + + return pd.to_datetime(result[0]) if result[0] else None + + def get_earliest_timestamp(self, symbol: str, timeframe: str) -> Optional[pd.Timestamp]: + """Get the earliest timestamp for a symbol/timeframe.""" + with self.get_connection() as conn: + cursor = conn.cursor() + cursor.execute(""" + SELECT MIN(timestamp) + FROM ohlcv + WHERE symbol = ? AND timeframe = ? + """, (symbol, timeframe)) + result = cursor.fetchone() + + return pd.to_datetime(result[0]) if result[0] else None diff --git a/data/fetcher.py b/data/fetcher.py index e33827c..62d371a 100644 --- a/data/fetcher.py +++ b/data/fetcher.py @@ -10,6 +10,8 @@ from binance.client import Client import pytz from dotenv import load_dotenv +from data.database import DatabaseManager + # 加载环境变量 load_dotenv() @@ -23,10 +25,10 @@ class BinanceDataFetcher: } def __init__(self, cache_dir: str = "./data/cache"): - """初始化Binance客户端和缓存目录 - + """初始化Binance客户端和数据库管理器 + Args: - cache_dir: 缓存下载数据的目录 + cache_dir: 缓存/数据库目录 """ # 配置代理 proxies = None @@ -58,8 +60,8 @@ class BinanceDataFetcher: print(" 3. 或者使用已缓存的数据") self.client = None - self.cache_dir = Path(cache_dir) - self.cache_dir.mkdir(parents=True, exist_ok=True) + # 初始化数据库管理器 + self.db_manager = DatabaseManager(os.path.join(cache_dir, "market_data.db")) def fetch_ohlcv( self, @@ -81,21 +83,7 @@ class BinanceDataFetcher: Returns: 包含以下列的DataFrame: timestamp, open, high, low, close, volume """ - # 生成缓存文件名 - cache_file = self._get_cache_filename(symbol, timeframe, start_date, end_date) - - # 检查缓存 - if use_cache and cache_file.exists(): - print(f"从缓存加载数据：{cache_file}") - return pd.read_csv(cache_file, parse_dates=['timestamp']) - - # 检查客户端是否可用 - if self.client is None: - raise ConnectionError( - "Binance客户端不可用。请使用VPN或使用已缓存的数据。" - ) - - # 设置默认日期 + # 设置日期范围 if end_date is None: end_dt = datetime.now(pytz.UTC) else: @@ -106,6 +94,34 @@ class BinanceDataFetcher: else: start_dt = datetime.strptime(start_date, '%Y-%m-%d').replace(tzinfo=pytz.UTC) + start_str = start_dt.strftime('%Y-%m-%d') + end_str = end_dt.strftime('%Y-%m-%d') + + # 1. 尝试从数据库获取数据 + df = pd.DataFrame() + if use_cache: + df = self.db_manager.get_ohlcv(symbol, timeframe, start_str, end_str) + + # 检查数据是否足够 + if not df.empty: + min_ts = df['timestamp'].min().replace(tzinfo=pytz.UTC) + max_ts = df['timestamp'].max().replace(tzinfo=pytz.UTC) + + # 简单判断覆盖范围 (容忍少量缺失) + if min_ts <= start_dt + timedelta(days=2) and max_ts >= end_dt - timedelta(days=2): + print(f"从数据库加载数据：{symbol} {timeframe} ({len(df)} rows)") + return df + + # 2. 从API获取数据 + if self.client is None: + if not df.empty: + print(f"⚠ API不可用，返回数据库现有数据 ({len(df)} rows)") + return df + + raise ConnectionError( + "Binance客户端不可用。请使用VPN或确保数据库有数据。" + ) + # Binance使用USDT交易对 pair = f"{symbol}USDT" interval = self.INTERVALS[timeframe] @@ -113,53 +129,47 @@ class BinanceDataFetcher: print(f"获取 {symbol} {timeframe} 数据，时间范围：{start_dt.date()} 至 {end_dt.date()}...") # 获取K线数据 - klines = self.client.get_historical_klines( - symbol=pair, - interval=interval, - start_str=str(int(start_dt.timestamp() * 1000)), - end_str=str(int(end_dt.timestamp() * 1000)) - ) + try: + klines = self.client.get_historical_klines( + symbol=pair, + interval=interval, + start_str=str(int(start_dt.timestamp() * 1000)), + end_str=str(int(end_dt.timestamp() * 1000)) + ) + except Exception as e: + if not df.empty: + print(f"⚠ API请求失败 ({e})，返回数据库现有数据") + return df + raise e if not klines: - raise ValueError(f"未获取到 {symbol} {timeframe} 的数据") + if not df.empty: + return df + raise ValueError(f"未获取到 {symbol} {timeframe} 的数据") # 转换为DataFrame - df = pd.DataFrame(klines, columns=[ + new_df = pd.DataFrame(klines, columns=[ 'timestamp', 'open', 'high', 'low', 'close', 'volume', 'close_time', 'quote_volume', 'trades', 'taker_buy_base', 'taker_buy_quote', 'ignore' ]) # 选择并转换列 - df = df[['timestamp', 'open', 'high', 'low', 'close', 'volume']].copy() - df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms') + new_df = new_df[['timestamp', 'open', 'high', 'low', 'close', 'volume']].copy() + new_df['timestamp'] = pd.to_datetime(new_df['timestamp'], unit='ms') for col in ['open', 'high', 'low', 'close', 'volume']: - df[col] = df[col].astype(float) + new_df[col] = new_df[col].astype(float) # 按时间排序 - df = df.sort_values('timestamp').reset_index(drop=True) + new_df = new_df.sort_values('timestamp').reset_index(drop=True) - print(f"已获取 {len(df)} 行数据，时间范围：{df['timestamp'].min()} 至 {df['timestamp'].max()}") + print(f"已获取 {len(new_df)} 行数据") - # 保存到缓存 - df.to_csv(cache_file, index=False) - print(f"已缓存数据至：{cache_file}") - - return df - - def _get_cache_filename( - self, - symbol: str, - timeframe: str, - start_date: Optional[str], - end_date: Optional[str] - ) -> Path: - """根据参数生成缓存文件名""" - start = start_date or "1y" - end = end_date or "now" - filename = f"{symbol}_{timeframe}_{start}_{end}.csv" - return self.cache_dir / filename + # 保存到数据库 + self.db_manager.save_ohlcv(new_df, symbol, timeframe) + + return new_df def get_latest_price(self, symbol: Literal['BTC', 'ETH', 'SOL']) -> float: """获取符号的当前价格 @@ -187,5 +197,8 @@ if __name__ == "__main__": print(f"日期范围：{df['timestamp'].min()} 至 {df['timestamp'].max()}") # 获取当前价格 - current_price = fetcher.get_latest_price('BTC') - print(f"\n当前BTC价格：${current_price:,.2f}") + try: + current_price = fetcher.get_latest_price('BTC') + print(f"\n当前BTC价格：${current_price:,.2f}") + except: + print("\n无法获取当前价格") diff --git a/ensemble_three_models.py b/ensemble_three_models.py index ae8903a..633d9fb 100644 --- a/ensemble_three_models.py +++ b/ensemble_three_models.py @@ -205,12 +205,13 @@ def main(): # 还原真实价格 nbeats_pred_prices = nbeats_preprocessor.inverse_transform_price(nbeats_pred_scaled) - # 【修改】获取当前K线的open价格（未来可交易价格）用于判断方向 - # 第i个样本对应的是第168+i根K线，其open价格是真正可以交易的价格 - current_open_prices = df_nbeats['open'].iloc[168:168+len(nbeats_pred_prices)].values - nbeats_preds = (nbeats_pred_prices > current_open_prices).astype(int) # 1=涨, 0=跌 + # 【修复】使用当前K线的close价格判断涨跌（与标签定义一致） + # 第i个样本对应的是第168+i根K线，其close价格是当前收盘价 + # 标签定义: label = (future_close > current_close) + current_close_prices = df_nbeats['close'].iloc[168:168+len(nbeats_pred_prices)].values + nbeats_preds = (nbeats_pred_prices > current_close_prices).astype(int) # 1=涨, 0=跌 - print(f"✓ N-BEATS预测完成 (使用当前K线open价格判断)") + print(f"✓ N-BEATS预测完成 (使用当前K线close价格判断)") print(f" 预测上涨: {(nbeats_preds==1).sum()} ({(nbeats_preds==1).sum()/len(nbeats_preds)*100:.1f}%)") print(f" 预测下跌: {(nbeats_preds==0).sum()} ({(nbeats_preds==0).sum()/len(nbeats_preds)*100:.1f}%)") @@ -223,7 +224,6 @@ def main(): # 从df_2026获取价格数据 all_prices = df_2026['close'].values - all_open_prices = df_2026['open'].values # 为每个样本创建预测 for i in range(min_samples): @@ -235,12 +235,12 @@ def main(): if end_idx > len(all_prices): # 如果超出范围，使用最后512个价格 context_data = all_prices[-context_length:].reshape(1, -1, 1) - # 【修改】使用对应K线的open价格作为当前可交易价格 - current_open_price = all_open_prices[min(168 + i, len(all_open_prices) - 1)] + # 【修复】使用对应K线的close价格作为当前收盘价 + current_close_price = all_prices[min(168 + i, len(all_prices) - 1)] else: context_data = all_prices[start_idx:end_idx].reshape(1, -1, 1) - # 【修改】使用第end_idx根K线的open价格（即当前K线的开盘价） - current_open_price = all_open_prices[end_idx] if end_idx < len(all_open_prices) else all_open_prices[-1] + # 【修复】使用第168+i根K线的close价格（即当前K线的收盘价） + current_close_price = all_prices[168 + i] if 168 + i < len(all_prices) else all_prices[-1] context_tensor = torch.FloatTensor(context_data).to(device) @@ -249,12 +249,12 @@ def main(): # 预测第一个时间步的值 pred_price = output.prediction_outputs.cpu().numpy()[0, 0, 0] - # 【修改】判断方向: 预测价格 vs 当前K线open价格 - ttm_preds_list.append(1 if pred_price > current_open_price else 0) + # 【修复】判断方向: 预测价格 vs 当前K线close价格（与标签定义一致） + ttm_preds_list.append(1 if pred_price > current_close_price else 0) ttm_preds = np.array(ttm_preds_list) - print(f"✓ TTM-R2预测完成 (使用当前K线open价格判断)") + print(f"✓ TTM-R2预测完成 (使用当前K线close价格判断)") print(f" 预测上涨: {(ttm_preds==1).sum()} ({(ttm_preds==1).sum()/len(ttm_preds)*100:.1f}%)") print(f" 预测下跌: {(ttm_preds==0).sum()} ({(ttm_preds==0).sum()/len(ttm_preds)*100:.1f}%)") @@ -358,38 +358,22 @@ def main(): print(f"\n集成策略性能:") print(f" 双模型交集 (DLinear+NBEATS): {two_model_acc*100:.2f}% (覆盖率: {two_model_valid.sum()/len(dlinear_preds)*100:.1f}%)") + print(f" 三模型多数投票: {ensemble_accuracy*100:.2f}% (覆盖率: 100%)") - if len(valid_preds) > 0: - print(f" 三模型交集 (ALL): {ensemble_accuracy*100:.2f}% (覆盖率: {valid_mask.sum()/len(dlinear_preds)*100:.1f}%)") + # 性能评估 + print("\n" + "="*80) + print("🎯 策略评估") + print("="*80) - # 性能评估 - print("\n" + "="*80) - print("🎯 策略评估") - print("="*80) + best_single = max(dlinear_acc, nbeats_acc, ttm_acc) - best_single = max(dlinear_acc, nbeats_acc, ttm_acc) - - if ensemble_accuracy > best_single: - improvement = ensemble_accuracy - best_single - print(f"\n ✅ 三模型交集优于所有单模型!") - print(f" 准确率提升: {improvement*100:.2f}个百分点") - print(f" 代价: 弃权{(~valid_mask).sum()}次预测 ({(~valid_mask).sum()/len(dlinear_preds)*100:.1f}%)") - elif ensemble_accuracy > two_model_acc: - improvement = ensemble_accuracy - two_model_acc - print(f"\n ✅ 三模型交集优于双模型交集!") - print(f" 准确率提升: {improvement*100:.2f}个百分点") - print(f" 但覆盖率下降: {(two_model_valid.sum() - valid_mask.sum())/len(dlinear_preds)*100:.1f}%") - else: - print(f"\n ⚠️ 三模型交集未能提升准确率") - print(f" 可能原因: TTM-R2表现太差({ttm_acc*100:.2f}%)，拖累了整体性能") - print(f" 建议: 使用双模型交集(DLinear+NBEATS)") - - # 实际收益分析 - print(f"\n💰 实际收益分析:") - print(f" 如果只在三个模型都同意时交易:") - print(f" - 交易次数: {valid_mask.sum()} ({valid_mask.sum()/len(dlinear_preds)*100:.1f}%)") - print(f" - 准确率: {ensemble_accuracy*100:.2f}%") - print(f" - 避免交易: {(~valid_mask).sum()}次 (规避不确定性)") + if ensemble_accuracy > best_single: + improvement = ensemble_accuracy - best_single + print(f"\n ✅ 三模型多数投票优于所有单模型!") + print(f" 准确率提升: {improvement*100:.2f}个百分点") + else: + print(f"\n ⚠️ 三模型多数投票未能超越最佳单模型") + print(f" 可能原因: 某个模型表现太差，拖累整体性能") print("\n" + "="*80) print("✓ 三模型集成预测分析完成!") diff --git a/scripts/generate_sample_data.py b/scripts/generate_sample_data.py index 837efa8..6e58d62 100644 --- a/scripts/generate_sample_data.py +++ b/scripts/generate_sample_data.py @@ -1,10 +1,17 @@ """生成测试数据以便在无法连接Binance时使用""" +import sys +import os +from pathlib import Path + +# 添加项目根目录 to path +sys.path.append(str(Path(__file__).resolve().parent.parent)) + import pandas as pd import numpy as np from datetime import datetime, timedelta -from pathlib import Path +from data.database import DatabaseManager def generate_sample_data( symbol: str = 'BTC', @@ -76,14 +83,13 @@ def generate_sample_data( df = pd.DataFrame(data) - # 保存到缓存 - cache_dir = Path('./data/cache') - cache_dir.mkdir(parents=True, exist_ok=True) + # 保存到数据库 + # 注意：我们这里假设DatabaseManager的默认路径在脚本运行时是相对于CWD的 + # 如果从根目录运行，./data/cache/market_data.db 是正确的 + db_manager = DatabaseManager() + db_manager.save_ohlcv(df, symbol, timeframe) - cache_file = cache_dir / f"{symbol}_{timeframe}_1y_now.csv" - df.to_csv(cache_file, index=False) - - print(f"✓ 已生成 {len(df)} 行模拟数据并保存到：{cache_file}") + print(f"✓ 已生成 {len(df)} 行模拟数据并保存到数据库") return df diff --git a/scripts/migrate_csv_to_db.py b/scripts/migrate_csv_to_db.py new file mode 100644 index 0000000..621c630 --- /dev/null +++ b/scripts/migrate_csv_to_db.py @@ -0,0 +1,67 @@ +"""Migrate existing CSV cache data to SQLite database.""" + +import sys +import os +from pathlib import Path +import re + +# Add project root to path +sys.path.append(str(Path(__file__).resolve().parent.parent)) + +import pandas as pd +from data.database import DatabaseManager + +def migrate(): + cache_dir = Path("./data/cache") + db_path = cache_dir / "market_data.db" + + print(f"Initializing database at {db_path}...") + db_manager = DatabaseManager(str(db_path)) + + # Find all CSV files + csv_files = list(cache_dir.glob("*.csv")) + print(f"Found {len(csv_files)} CSV files to migrate.") + + success_count = 0 + fail_count = 0 + + for csv_file in csv_files: + try: + filename = csv_file.name + print(f"Processing {filename}...") + + # Parse filename + parts = filename.split('_') + if len(parts) < 2: + print(f" ⚠ Skipping {filename}: Invalid naming format") + continue + + symbol = parts[0] + timeframe = parts[1] + + # Validate symbol and timeframe (basic check) + if timeframe not in ['1h', '15m']: + # Relaxed check, just warning + print(f" ℹ Note: Unfamiliar timeframe {timeframe} in {filename}") + + # Read CSV + df = pd.read_csv(csv_file) + if 'timestamp' not in df.columns: + print(f" ⚠ Skipping {filename}: No timestamp column") + continue + + # Save to DB + db_manager.save_ohlcv(df, symbol, timeframe) + print(f" ✓ Imported {len(df)} records for {symbol} {timeframe}") + success_count += 1 + + except Exception as e: + print(f" ❌ Error processing {csv_file.name}: {e}") + fail_count += 1 + + print("\nMigration complete!") + print(f"Successfully migrated: {success_count} files") + print(f"Failed: {fail_count} files") + +if __name__ == "__main__": + migrate() diff --git a/trading_log.json b/trading_log.json new file mode 100644 index 0000000..a1afb17 --- /dev/null +++ b/trading_log.json @@ -0,0 +1,13 @@ +[ + { + "timestamp": "2026-02-03T21:18:33.773774", + "market": "Bitcoin Up or Down - February 3, 9AM ET", + "prediction": "DOWN", + "success": false, + "order": null, + "error": "价格$0.51超过限制$0.50", + "side": "DOWN", + "price": 0.505, + "probability": 0.5563886165618896 + } +] \ No newline at end of file