添加数据库文件

api/__init__.py

@@ -0,0 +1 @@
+# api模块初始化

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()

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()

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/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

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线数据
+        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:
+             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()
+        new_df = new_df[['timestamp', 'open', 'high', 'low', 'close', 'volume']].copy()
-        df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
+        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)
+        self.db_manager.save_ohlcv(new_df, symbol, timeframe)
-        print(f"已缓存数据至：{cache_file}")
         
-        return df
+        return new_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
 
     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()}")
 
     # 获取当前价格
+    try:
         current_price = fetcher.get_latest_price('BTC')
         print(f"\n当前BTC价格：${current_price:,.2f}")
+    except:
+        print("\n无法获取当前价格")

ensemble_three_models.py

@@ -205,12 +205,13 @@ def main():
     # 还原真实价格
     nbeats_pred_prices = nbeats_preprocessor.inverse_transform_price(nbeats_pred_scaled)
 
-    # 【修改】获取当前K线的open价格（未来可交易价格）用于判断方向
+    # 【修复】使用当前K线的close价格判断涨跌（与标签定义一致）
-    # 第i个样本对应的是第168+i根K线，其open价格是真正可以交易的价格
+    # 第i个样本对应的是第168+i根K线，其close价格是当前收盘价
-    current_open_prices = df_nbeats['open'].iloc[168:168+len(nbeats_pred_prices)].values
+    # 标签定义: label = (future_close > current_close) 
-    nbeats_preds = (nbeats_pred_prices > current_open_prices).astype(int)  # 1=涨, 0=跌
+    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价格作为当前可交易价格
+            # 【修复】使用对应K线的close价格作为当前收盘价
-            current_open_price = all_open_prices[min(168 + i, len(all_open_prices) - 1)]
+            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线的开盘价）
+            # 【修复】使用第168+i根K线的close价格（即当前K线的收盘价）
-            current_open_price = all_open_prices[end_idx] if end_idx < len(all_open_prices) else all_open_prices[-1]
+            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价格
+        # 【修复】判断方向: 预测价格 vs 当前K线close价格（与标签定义一致）
-        ttm_preds_list.append(1 if pred_price > current_open_price else 0)
+        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,9 +358,7 @@ 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)
@@ -371,25 +369,11 @@ def main():
 
     if ensemble_accuracy > best_single:
         improvement = ensemble_accuracy - best_single
-            print(f"\n  ✅ 三模型交集优于所有单模型!")
+        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"\n  ⚠️  三模型多数投票未能超越最佳单模型")
-            print(f"     可能原因: TTM-R2表现太差({ttm_acc*100:.2f}%)，拖累了整体性能")
+        print(f"     可能原因: 某个模型表现太差，拖累整体性能")
-            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()}次 (规避不确定性)")
 
     print("\n" + "="*80)
     print("✓ 三模型集成预测分析完成!")

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')
+    # 注意：我们这里假设DatabaseManager的默认路径在脚本运行时是相对于CWD的
-    cache_dir.mkdir(parents=True, exist_ok=True)
+    # 如果从根目录运行，./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"
+    print(f"✓ 已生成 {len(df)} 行模拟数据并保存到数据库")
-    df.to_csv(cache_file, index=False)
-
-    print(f"✓ 已生成 {len(df)} 行模拟数据并保存到：{cache_file}")
     return df
 
 

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()

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
+  }
+]