diff --git a/api/polymarket_client.py b/api/polymarket_client.py index 185341f..f363ac1 100644 --- a/api/polymarket_client.py +++ b/api/polymarket_client.py @@ -35,6 +35,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( diff --git a/data/fetcher.py b/data/fetcher.py index 21aa215..f57da7d 100644 --- a/data/fetcher.py +++ b/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 - use_proxy = os.getenv('USE_PROXY', 'false').lower() == 'true' - - if use_proxy: - socks5_proxy = os.getenv('SOCKS5_PROXY', 'socks5://127.0.0.1:7890') - proxies = { - 'http': socks5_proxy, - 'https': socks5_proxy - } - print(f"✓ 使用SOCKS5代理: {socks5_proxy}") + socks5_proxy = os.getenv('SOCKS5_PROXY', 'socks5://127.0.0.1:7890') + _apply_global_proxy(socks5_proxy) + proxies = { + 'http': socks5_proxy, + 'https': socks5_proxy + } + logger.info(f"强制使用SOCKS5代理: {socks5_proxy}") try: - # 初始化Binance客户端（带代理） - if proxies: - self.client = Client(requests_params={'proxies': proxies}) - else: - self.client = Client() + # 初始化Binance客户端（强制走代理） + self.client = Client(requests_params={'proxies': proxies}) # 测试连接 self.client.ping() - print("✓ 已成功连接到Binance API") + logger.info("已成功连接到Binance API") except Exception as e: - print(f"⚠ Binance API连接失败: {e}") - print("提示：") - print(" 1. 检查代理设置 (.env文件)") - print(" 2. 确保代理服务正在运行") - print(" 3. 或者使用已缓存的数据") + logger.warning(f"Binance API连接失败: {e}") + logger.warning("提示: 1) 检查代理设置 2) 确保代理服务运行 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)), - end_str=str(int(end_dt.timestamp() * 1000)) + start_ts_ms=int(start_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']) diff --git a/requirements.txt b/requirements.txt index 758840e..6b1b937 100644 --- a/requirements.txt +++ b/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 # ---------------------------------------------- # 任务调度 diff --git a/scripts/hourly_fetch_btc_1h.py b/scripts/hourly_fetch_btc_1h.py new file mode 100644 index 0000000..7820858 --- /dev/null +++ b/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() diff --git a/trader/auto_trade_polymarket.py b/trader/auto_trade_polymarket.py index 662301e..036a69f 100644 --- a/trader/auto_trade_polymarket.py +++ b/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 - backupCount=5, + when='midnight', + 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, diff --git a/utils/send_msg.py b/utils/send_msg.py index 5cdeeb7..2e39f1f 100644 --- a/utils/send_msg.py +++ b/utils/send_msg.py @@ -8,9 +8,12 @@ class LarkNotifier: self.url = "https://zhiyinzhe.xyz/sendMsg/api/send" self.bot_name = bot_name self.receive_id = receive_id - - # 使用 Session 可以复用底层 TCP 连接，提升发送效率 - self.session = requests.Session() + + # 代理固定为本地 7898 端口 + 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 - response = self.session.post(self.url, data=payload, timeout=5) + # 每次请求直接调用 requests.post，不复用 Session + response = requests.post(self.url, data=payload, timeout=5, proxies=self.proxies) response.raise_for_status() print(f"告警发送成功: {title}") except Exception as e: