merge: feature/websocket-kline-daily-log into main

api/polymarket_client.py

@@ -36,6 +36,16 @@ class PolymarketClient:
     GAMMA_API = "https://gamma-api.polymarket.com"
     CLOB_API = "https://clob.polymarket.com"
 
+    @staticmethod
+    def _apply_global_proxy(proxy_url: str) -> None:
+        """统一设置代理环境变量，确保第三方网络库也走代理。"""
+        os.environ["HTTP_PROXY"] = proxy_url
+        os.environ["HTTPS_PROXY"] = proxy_url
+        os.environ["ALL_PROXY"] = proxy_url
+        os.environ["http_proxy"] = proxy_url
+        os.environ["https_proxy"] = proxy_url
+        os.environ["all_proxy"] = proxy_url
+    
     def __init__(self, private_key: Optional[str] = None):
         """
         初始化客户端
@@ -46,7 +56,8 @@ class PolymarketClient:
         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
+        proxy = os.getenv("SOCKS5_PROXY", "socks5://127.0.0.1:7890")
+        self._apply_global_proxy(proxy)
         
         # HTTP客户端 - 禁用压缩避免brotli问题
         self.http = httpx.Client(

data/fetcher.py

@@ -1,15 +1,19 @@
 """从Binance获取加密货币OHLCV数据"""
 
 import os
+import json
 import time
 from datetime import datetime, timedelta
 from pathlib import Path
-from typing import Optional, Literal
+from typing import Optional, Literal, Any
+import logging
+from urllib.parse import urlparse
 
 import pandas as pd
 from binance.client import Client
 import pytz
 from dotenv import load_dotenv
+from websocket import create_connection
 
 from data.database import DatabaseManager
 from utils.send_msg import lark_notifier
@@ -18,6 +22,19 @@ from utils.send_msg import lark_notifier
 load_dotenv()
 
 
+logger = logging.getLogger("polymarket_trader")
+
+
+def _apply_global_proxy(proxy_url: str) -> None:
+    """统一设置代理环境变量，确保第三方网络库也走代理。"""
+    os.environ['HTTP_PROXY'] = proxy_url
+    os.environ['HTTPS_PROXY'] = proxy_url
+    os.environ['ALL_PROXY'] = proxy_url
+    os.environ['http_proxy'] = proxy_url
+    os.environ['https_proxy'] = proxy_url
+    os.environ['all_proxy'] = proxy_url
+
+
 class BinanceDataFetcher:
     """从Binance公开API获取和缓存OHLCV数据"""
 
@@ -26,6 +43,13 @@ class BinanceDataFetcher:
         '15m': Client.KLINE_INTERVAL_15MINUTE,
     }
 
+    INTERVAL_MS = {
+        '1h': 60 * 60 * 1000,
+        '15m': 15 * 60 * 1000,
+    }
+
+    WS_API_ENDPOINT = "wss://ws-api.binance.com/ws-api/v3"
+
     def __init__(self, cache_dir: str = "./data/cache"):
         """初始化Binance客户端和数据库管理器
         
@@ -33,33 +57,24 @@ class BinanceDataFetcher:
             cache_dir: 缓存/数据库目录
         """
         # 配置代理
-        proxies = None
+        socks5_proxy = os.getenv('SOCKS5_PROXY', 'socks5://127.0.0.1:7890')
-        use_proxy = os.getenv('USE_PROXY', 'false').lower() == 'true'
+        _apply_global_proxy(socks5_proxy)
-
+        proxies = {
-        if use_proxy:
+            'http': socks5_proxy,
-            socks5_proxy = os.getenv('SOCKS5_PROXY', 'socks5://127.0.0.1:7890')
+            'https': socks5_proxy
-            proxies = {
+        }
-                'http': socks5_proxy,
+        logger.info(f"强制使用SOCKS5代理: {socks5_proxy}")
-                'https': socks5_proxy
-            }
-            print(f"✓ 使用SOCKS5代理: {socks5_proxy}")
 
         try:
-            # 初始化Binance客户端（带代理）
+            # 初始化Binance客户端（强制走代理）
-            if proxies:
+            self.client = Client(requests_params={'proxies': proxies})
-                self.client = Client(requests_params={'proxies': proxies})
-            else:
-                self.client = Client()
 
             # 测试连接
             self.client.ping()
-            print("✓ 已成功连接到Binance API")
+            logger.info("已成功连接到Binance API")
         except Exception as e:
-            print(f"⚠ Binance API连接失败: {e}")
+            logger.warning(f"Binance API连接失败: {e}")
-            print("提示：")
+            logger.warning("提示: 1) 检查代理设置 2) 确保代理服务运行 3) 或者使用缓存数据")
-            print("  1. 检查代理设置 (.env文件)")
-            print("  2. 确保代理服务正在运行")
-            print("  3. 或者使用已缓存的数据")
             self.client = None
 
         # 初始化数据库管理器
@@ -112,13 +127,13 @@ class BinanceDataFetcher:
                 
                 # 简单判断覆盖范围 (容忍少量缺失)
                 if min_ts <= start_dt + timedelta(days=2) and max_ts >= end_dt - timedelta(days=2):
-                    print(f"从数据库加载数据：{symbol} {timeframe} ({len(df)} rows)")
+                    logger.info(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)")
+                 logger.warning(f"API不可用，返回数据库现有数据 ({len(df)} rows)")
                  return df
             
             raise ConnectionError(
@@ -128,32 +143,35 @@ class BinanceDataFetcher:
         # Binance使用USDT交易对
         pair = f"{symbol}USDT"
         interval = self.INTERVALS[timeframe]
+        interval_ms = self.INTERVAL_MS[timeframe]
 
-        print(f"获取 {symbol} {timeframe} 数据，时间范围：{start_dt.date()} 至 {end_dt.date()}...")
+        logger.info(f"获取 {symbol} {timeframe} 数据，时间范围: {start_dt.date()} 至 {end_dt.date()}")
 
         klines = []
         for attempt in range(max_retries):
             try:
-                klines = self.client.get_historical_klines(
+                # 使用 Binance WebSocket API 的 klines 方法获取历史K线
+                klines = self._get_historical_klines_via_ws(
                     symbol=pair,
                     interval=interval,
-                    start_str=str(int(start_dt.timestamp() * 1000)),
+                    start_ts_ms=int(start_dt.timestamp() * 1000),
-                    end_str=str(int(end_dt.timestamp() * 1000))
+                    end_ts_ms=int(end_dt.timestamp() * 1000),
+                    interval_ms=interval_ms
                 )
                 # 如果成功获取，跳出循环
                 break
             except Exception as e:
                 # 如果是最后一次尝试，且依然失败
                 if attempt == max_retries - 1:
-                    print(f"❌ 第 {attempt + 1} 次尝试失败，已达到最大重试次数。错误: {e}")
+                    logger.error(f"第 {attempt + 1} 次尝试失败，已达到最大重试次数。错误: {e}")
                     if not df.empty:
-                        print(f"⚠ API请求最终失败，返回数据库现有数据")
+                        logger.warning("WebSocket请求最终失败，返回数据库现有数据")
                         return df
                     raise e  # 如果没有缓存数据，抛出异常
                 
                 # 计算等待时间 (指数退避: 2s, 4s, 8s...)
                 wait_time = 2 * (2 ** attempt) 
-                print(f"⚠ 请求失败 (尝试 {attempt + 1}/{max_retries})，{wait_time}秒后重试... 错误: {e}")
+                logger.warning(f"请求失败 (尝试 {attempt + 1}/{max_retries})，{wait_time}秒后重试。错误: {e}")
                 time.sleep(wait_time)
 
         if not klines:
@@ -179,13 +197,94 @@ class BinanceDataFetcher:
         # 按时间排序
         new_df = new_df.sort_values('timestamp').reset_index(drop=True)
 
-        print(f"已获取 {len(new_df)} 行数据")
+        logger.info(f"已获取 {len(new_df)} 行数据")
 
         # 保存到数据库
         self.db_manager.save_ohlcv(new_df, symbol, timeframe)
         
         return new_df
 
+    def _get_historical_klines_via_ws(
+        self,
+        symbol: str,
+        interval: str,
+        start_ts_ms: int,
+        end_ts_ms: int,
+        interval_ms: int,
+        limit: int = 1000
+    ):
+        """通过 Binance WebSocket API 分页拉取历史K线。"""
+        if end_ts_ms <= start_ts_ms:
+            return []
+
+        all_klines = []
+        request_id = 1
+        current_start = start_ts_ms
+
+        # WebSocket API 单次最多返回 1000 条，这里按窗口分页拉取。
+        while current_start < end_ts_ms:
+            max_window_end = current_start + (limit * interval_ms)
+            current_end = min(end_ts_ms, max_window_end)
+
+            payload = {
+                "id": request_id,
+                "method": "klines",
+                "params": {
+                    "symbol": symbol,
+                    "interval": interval,
+                    "startTime": current_start,
+                    "endTime": current_end,
+                    "limit": limit
+                }
+            }
+
+            response_data = self._ws_request(payload)
+            batch = response_data.get("result", [])
+            if not batch:
+                break
+
+            all_klines.extend(batch)
+            last_open_time = int(batch[-1][0])
+            next_start = last_open_time + interval_ms
+            if next_start <= current_start:
+                break
+            current_start = next_start
+            request_id += 1
+
+        return all_klines
+
+    def _ws_request(self, payload: dict) -> dict:
+        """发送单次 WebSocket API 请求并返回 JSON 响应。"""
+        ws_kwargs: dict[str, Any] = {
+            "timeout": 30
+        }
+
+        socks5_proxy = os.getenv('SOCKS5_PROXY', 'socks5://127.0.0.1:7890')
+        parsed = urlparse(socks5_proxy)
+        if parsed.hostname and parsed.port:
+            scheme = (parsed.scheme or 'socks5').lower()
+            if scheme not in {'socks4', 'socks5', 'http'}:
+                scheme = 'socks5'
+            if scheme == 'socks5h':
+                scheme = 'socks5'
+            ws_kwargs.update({
+                "http_proxy_host": parsed.hostname,
+                "http_proxy_port": parsed.port,
+                "proxy_type": scheme
+            })
+
+        ws = create_connection(self.WS_API_ENDPOINT, **ws_kwargs)
+        try:
+            ws.send(json.dumps(payload))
+            raw = ws.recv()
+            data = json.loads(raw)
+            status = data.get("status", 200)
+            if status != 200:
+                raise RuntimeError(f"Binance WS API返回异常状态: {data}")
+            return data
+        finally:
+            ws.close()
+
     def get_latest_price(self, symbol: Literal['BTC', 'ETH', 'SOL']) -> float:
         """获取符号的当前价格
 
@@ -195,6 +294,9 @@ class BinanceDataFetcher:
         Returns:
             当前USDT价格
         """
+        if self.client is None:
+            raise ConnectionError("Binance客户端不可用，无法获取最新价格")
+
         pair = f"{symbol}USDT"
         ticker = self.client.get_symbol_ticker(symbol=pair)
         return float(ticker['price'])

requirements.txt

@@ -46,6 +46,9 @@ httpx>=0.25.0
 # 数据源
 # ----------------------------------------------
 python-binance>=1.0.19
+websocket-client>=1.8.0
+python-socks>=2.5.0
+PySocks>=1.7.1
 
 # ----------------------------------------------
 # 任务调度

scripts/hourly_fetch_btc_1h.py

@@ -0,0 +1,112 @@
+#!/usr/bin/env python3
+"""整点后5秒抓取 BTC 1h K线（至少512根，且包含最新时间）"""
+
+import os
+import sys
+import time
+from datetime import datetime, timedelta
+
+import pandas as pd
+import pytz
+
+# 直接运行脚本时，确保可以从项目根目录导入 data 包
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from data.fetcher import BinanceDataFetcher
+
+
+def get_next_run_time(delay_seconds: int = 5) -> datetime:
+    """计算下一次执行时间：下一个整点 + delay_seconds。"""
+    now_utc = datetime.now(pytz.UTC)
+    next_hour = (now_utc + timedelta(hours=1)).replace(minute=0, second=0, microsecond=0)
+    return next_hour + timedelta(seconds=delay_seconds)
+
+
+def fetch_min_klines_with_latest(
+    fetcher: BinanceDataFetcher,
+    min_klines: int = 512,
+    latest_check_retries: int = 8,
+    latest_check_interval: int = 2,
+) -> pd.DataFrame:
+    """
+    拉取 BTC 1h K线，确保：
+    1) 数量不少于 min_klines
+    2) 包含当前小时的最新时间（UTC 整点）
+    """
+    now_utc = datetime.now(pytz.UTC)
+    expected_latest_ts = now_utc.replace(minute=0, second=0, microsecond=0).replace(tzinfo=None)
+
+    lookback_days = 35
+    for attempt in range(latest_check_retries):
+        start_date = (datetime.now(pytz.UTC) - timedelta(days=lookback_days)).strftime("%Y-%m-%d")
+
+        # 使用 WebSocket 拉取，避免仅命中缓存导致最新K线不更新。
+        df = fetcher.fetch_ohlcv(
+            symbol="BTC",
+            timeframe="1h",
+            start_date=start_date,
+            end_date=None,
+            use_cache=False,
+            max_retries=3,
+        )
+
+        if len(df) < min_klines:
+            lookback_days += 7
+            print(
+                f"[WARN] 当前仅获取 {len(df)} 根K线，小于 {min_klines}，扩大回看窗口到 {lookback_days} 天后重试。"
+            )
+            continue
+
+        latest_ts = df["timestamp"].max().to_pydatetime().replace(tzinfo=None)
+        if latest_ts >= expected_latest_ts:
+            result = df.tail(min_klines).copy().reset_index(drop=True)
+            print(
+                f"[OK] 获取成功: {len(result)} 根 | 最新时间: {latest_ts} UTC | 期望最新时间: {expected_latest_ts} UTC"
+            )
+            return result
+
+        if attempt < latest_check_retries - 1:
+            print(
+                f"[WAIT] 最新K线未到位，当前最新: {latest_ts} UTC，期望: {expected_latest_ts} UTC，"
+                f"{latest_check_interval}秒后重试({attempt + 1}/{latest_check_retries})。"
+            )
+            time.sleep(latest_check_interval)
+
+    raise RuntimeError("多次重试后仍未拿到包含最新时间的BTC 1h K线数据")
+
+
+def run_once(min_klines: int = 512) -> None:
+    """执行一次抓取任务。"""
+    fetcher = BinanceDataFetcher()
+    data = fetch_min_klines_with_latest(fetcher=fetcher, min_klines=min_klines)
+
+    print("-" * 80)
+    print(f"抓取完成，返回K线数量: {len(data)}")
+    print(f"时间范围: {data['timestamp'].min()} -> {data['timestamp'].max()} (UTC)")
+    print(data.tail(3)[["timestamp", "open", "high", "low", "close", "volume"]])
+    print("-" * 80)
+
+
+def run_forever(min_klines: int = 512, delay_seconds: int = 5) -> None:
+    """常驻运行：每个整点延迟 delay_seconds 后执行抓取。"""
+    while True:
+        next_run = get_next_run_time(delay_seconds=delay_seconds)
+        sleep_seconds = max(0.0, (next_run - datetime.now(pytz.UTC)).total_seconds())
+
+        print(f"[SCHEDULE] 当前UTC: {datetime.now(pytz.UTC)}，下次执行: {next_run}，等待 {sleep_seconds:.1f} 秒")
+        time.sleep(sleep_seconds)
+
+        try:
+            run_once(min_klines=min_klines)
+        except Exception as exc:
+            print(f"[ERROR] 本轮抓取失败: {exc}")
+
+
+def main() -> None:
+    """脚本入口。"""
+    # 默认常驻运行，如需只执行一次，可把这里改成 run_once(min_klines=512)
+    run_forever(min_klines=512, delay_seconds=5)
+
+
+if __name__ == "__main__":
+    main()

trader/auto_trade_polymarket.py

@@ -15,7 +15,7 @@ import argparse
 import json
 import time
 import logging
-from logging.handlers import RotatingFileHandler
+from logging.handlers import TimedRotatingFileHandler
 import schedule
 from datetime import datetime, timedelta
 from typing import Optional, Tuple, Dict, List
@@ -68,24 +68,20 @@ def setup_logging(log_dir: str = "logs", log_level: int = logging.INFO) -> loggi
         datefmt="%Y-%m-%d %H:%M:%S"
     )
     
-    # 文件handler (按大小轮转, 最大10MB, 保留5个备份)
+    # 文件handler (按天轮转, 保留最近30天)
     log_file = os.path.join(log_dir, "polymarket_trading.log")
-    file_handler = RotatingFileHandler(
+    file_handler = TimedRotatingFileHandler(
         log_file,
-        maxBytes=10*1024*1024,  # 10MB
+        when='midnight',
-        backupCount=5,
+        interval=1,
+        backupCount=30,
         encoding='utf-8'
     )
+    file_handler.suffix = "%Y-%m-%d"
     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
 
 
@@ -93,6 +89,16 @@ def setup_logging(log_dir: str = "logs", log_level: int = logging.INFO) -> loggi
 logger = setup_logging()
 
 
+def _apply_global_proxy(proxy_url: str) -> None:
+    """统一设置代理环境变量，确保第三方网络库也走代理。"""
+    os.environ["HTTP_PROXY"] = proxy_url
+    os.environ["HTTPS_PROXY"] = proxy_url
+    os.environ["ALL_PROXY"] = proxy_url
+    os.environ["http_proxy"] = proxy_url
+    os.environ["https_proxy"] = proxy_url
+    os.environ["all_proxy"] = proxy_url
+
+
 class PolymarketBTCTrader:
     """Polymarket BTC Hourly交易器 (多模型集成版)"""
     
@@ -119,7 +125,8 @@ class PolymarketBTCTrader:
         self.dry_run = dry_run
         
         # HTTP客户端
-        proxy = os.getenv("SOCKS5_PROXY") if os.getenv("USE_PROXY", "false").lower() == "true" else None
+        proxy = os.getenv("SOCKS5_PROXY", "socks5://127.0.0.1:7890")
+        _apply_global_proxy(proxy)
         self.http = httpx.Client(
             timeout=30.0,
             proxy=proxy,

utils/send_msg.py

@@ -9,8 +9,11 @@ class LarkNotifier:
         self.bot_name = bot_name
         self.receive_id = receive_id
 
-        # 使用 Session 可以复用底层 TCP 连接，提升发送效率
+        # 代理固定为本地 7898 端口
-        self.session = requests.Session()
+        self.proxies = {
+            "http": "socks5h://127.0.0.1:7898",
+            "https": "socks5h://127.0.0.1:7898"
+        }
 
     def send_alert(self, title: str, content: str):
         """
@@ -26,8 +29,8 @@ class LarkNotifier:
         }
         
         try:
-            # 使用 data=payload 发送 Form Data
+            # 每次请求直接调用 requests.post，不复用 Session
-            response = self.session.post(self.url, data=payload, timeout=5)
+            response = requests.post(self.url, data=payload, timeout=5, proxies=self.proxies)
             response.raise_for_status()
             print(f"告警发送成功: {title}")
         except Exception as e: