From db937fdcf925b84e42acbc0b0a1ff2cebb8002f9 Mon Sep 17 00:00:00 2001 From: crypto_predictor Date: Fri, 6 Feb 2026 15:48:45 +0800 Subject: [PATCH] feat: Add DB logging, price monitoring, and duplicate trade prevention --- auto_trade_polymarket.py | 116 +++++++++++++++++++++++++++++++++++---- data/trading_db.py | 109 ++++++++++++++++++++++++++++++++++++ 2 files changed, 215 insertions(+), 10 deletions(-) create mode 100644 data/trading_db.py diff --git a/auto_trade_polymarket.py b/auto_trade_polymarket.py index 48b8a32..df67bd9 100644 --- a/auto_trade_polymarket.py +++ b/auto_trade_polymarket.py @@ -34,6 +34,7 @@ import httpx sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from data.fetcher import BinanceDataFetcher +from data.trading_db import TradingDatabase load_dotenv() @@ -186,6 +187,9 @@ class PolymarketBTCTrader: # 数据获取器 self.fetcher = BinanceDataFetcher() + # 数据库 + self.db = TradingDatabase() + # CLOB客户端(用于下单) self.clob_client = None if self.private_key and not self.dry_run: @@ -229,6 +233,42 @@ class PolymarketBTCTrader: except Exception as e: logger.warning(f"⚠️ CLOB客户端初始化失败: {e}") self.clob_client = None + + def _log_to_db(self, result: Dict, market: Optional[Dict]): + """记录交易结果到数据库""" + try: + # 确定状态 + status = 'failure' + if self.dry_run: + status = 'test' + elif result.get('success'): + status = 'success' + elif result.get('error'): + status = 'failure' + elif result.get('prediction'): # 有预测但未交易（如市场未找到） + status = 'in_progress' # 或者 failure? 未找到市场算 failure + if result.get('error'): + status = 'failure' + + log_data = { + 'timestamp': result.get('timestamp'), + 'market_title': result.get('market'), + 'market_slug': market.get('slug') if market else result.get('market_slug'), + 'model_name': 'DLinear', + 'prediction': result.get('prediction'), + 'probability': result.get('probability'), + 'action': result.get('side'), + 'price': result.get('price'), + 'amount': result.get('amount'), + 'size': result.get('size'), + 'status': status, + 'error_message': result.get('error'), + 'raw_data': result + } + self.db.log_trade(log_data) + logger.info(f"💾 交易记录已保存到数据库") + except Exception as e: + logger.error(f"❌ 保存数据库失败: {e}") def generate_market_slug(self, target_time: Optional[datetime] = None) -> str: """ @@ -598,7 +638,12 @@ class PolymarketBTCTrader: return prediction, probability - def run_at_hour_start(self, delay_seconds: float = 2.0) -> Dict: + def run_once(self) -> Dict: + """运行一次交易流程（不等待整点）""" + logger.info("运行单次交易...") + return self.run_at_hour_start(delay_seconds=0) + + def run_at_hour_start(self, delay_seconds: float = 2.0, timeout_minutes: int = 50) -> Dict: """ 整点时执行的交易流程 @@ -606,11 +651,13 @@ class PolymarketBTCTrader: 2. 获取最新K线 3. DLinear预测当前整点的收盘涨跌 4. 获取对应的Polymarket市场 - 5. 检查价格是否≤max_price - 6. 下单 + 5. 循环检查价格: + - 如果价格 <= max_price: 下单并结束 + - 如果价格 > max_price: 等待并重试(直到超时) Args: delay_seconds: 整点后延迟秒数 + timeout_minutes: 价格等待超时时间(分钟) Returns: 执行结果 @@ -621,6 +668,7 @@ class PolymarketBTCTrader: logger.info(f" 模式: {'🔸 模拟' if self.dry_run else '💰 真实交易'}") logger.info(f" 最大下注: ${self.max_bet:.2f} USDC") logger.info(f" 最高价格: ${self.max_price:.2f}") + logger.info(f" 等待超时: {timeout_minutes}分钟") result = { 'timestamp': datetime.now().isoformat(), @@ -633,24 +681,75 @@ class PolymarketBTCTrader: result['prediction'] = 'UP' if prediction == 1 else 'DOWN' result['probability'] = probability if prediction == 1 else 1 - probability - # 2. 获取当前整点的市场(不是下一个整点) + # 2. 获取初始市场 market = self.get_current_hourly_market() if not market: result['error'] = '未找到市场' logger.error(f"❌ {result['error']}") + self._log_to_db(result, None) return result result['market'] = market['title'] + result['market_slug'] = market['slug'] if market['closed']: result['error'] = '市场已关闭' logger.error(f"❌ {result['error']}") + self._log_to_db(result, market) return result + + # 检查是否已交易过该市场 + check_status = 'test' if self.dry_run else 'success' + if self.db.check_existing_trade(market['slug'], status=check_status): + result['error'] = '该市场已存在成功交易记录' + logger.warning(f"⚠️ {result['error']}, 跳过交易") + return result + + # 3. 循环检查价格并交易 + start_time = datetime.now() + end_time = start_time + timedelta(minutes=timeout_minutes) - # 3. 执行交易 - trade_result = self.execute_trade(prediction, market) - result.update(trade_result) + logger.info(f"🔄 开始价格监控 (预测: {'UP' if prediction == 1 else 'DOWN'})...") + while datetime.now() < end_time: + # 刷新价格 + market = self.refresh_prices() + if not market: + logger.warning("无法获取市场价格，重试中...") + time.sleep(10) + continue + + # 检查价格 + price_idx = 0 if prediction == 1 else 1 + current_price = market['outcome_prices'][price_idx] + + logger.info(f" 当前价格: ${current_price:.3f} (目标: <= ${self.max_price:.2f})") + + if current_price <= self.max_price: + # 价格符合，执行交易 + logger.info("✅ 价格满足条件，正在下单...") + trade_result = self.execute_trade(prediction, market) + result.update(trade_result) + + # 记录数据库 + self._log_to_db(result, market) + + logger.info("=" * 60) + return result + else: + # 价格过高，等待 + wait_time = 30 # 等待30秒 + elapsed = int((datetime.now() - start_time).total_seconds()) + logger.info(f"⏳ 价格过高，继续等待... (已耗时: {elapsed}s)") + time.sleep(wait_time) + + # 超时 + result['error'] = f'等待价格超时 ({timeout_minutes}分钟)' + result['price'] = market['outcome_prices'][0] if prediction == 1 else market['outcome_prices'][1] + result['side'] = 'UP' if prediction == 1 else 'DOWN' + logger.error(f"❌ {result['error']}") + + self._log_to_db(result, market) logger.info("=" * 60) return result @@ -758,9 +857,6 @@ class PolymarketBTCTrader: # 启动时进行一次预测（不下单，仅显示） 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) diff --git a/data/trading_db.py b/data/trading_db.py new file mode 100644 index 0000000..a83438c --- /dev/null +++ b/data/trading_db.py @@ -0,0 +1,109 @@ +import sqlite3 +from datetime import datetime +import json +import os +from contextlib import contextmanager + +class TradingDatabase: + """交易数据库管理类 - 用于记录交易历史和预测结果""" + + def __init__(self, db_path="trading_history.db"): + self.db_path = db_path + self._init_db() + + @contextmanager + def get_connection(self): + """数据库连接上下文管理器""" + conn = sqlite3.connect(self.db_path) + try: + yield conn + finally: + conn.close() + + def _init_db(self): + """初始化数据库表结构""" + with self.get_connection() as conn: + cursor = conn.cursor() + # 创建交易记录表 + cursor.execute(''' + CREATE TABLE IF NOT EXISTS trades ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + timestamp TEXT, + market_title TEXT, + market_slug TEXT, + model_name TEXT, + prediction TEXT, + probability REAL, + action TEXT, + price REAL, + amount REAL, + size REAL, + status TEXT, + error_message TEXT, + raw_data TEXT, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + ''') + conn.commit() + + def log_trade(self, trade_data: dict): + """ + 记录交易信息 + + Args: + trade_data: 包含交易信息的字典，支持字段: + timestamp, market_title, market_slug, model_name, + prediction, probability, action, price, amount, size, + status, error_message, raw_data + """ + try: + with self.get_connection() as conn: + cursor = conn.cursor() + cursor.execute(''' + INSERT INTO trades ( + timestamp, market_title, market_slug, model_name, + prediction, probability, action, price, amount, size, + status, error_message, raw_data + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ''', ( + trade_data.get('timestamp', datetime.now().isoformat()), + trade_data.get('market_title'), + trade_data.get('market_slug'), + trade_data.get('model_name'), + trade_data.get('prediction'), + trade_data.get('probability'), + trade_data.get('action'), + trade_data.get('price'), + trade_data.get('amount'), + trade_data.get('size'), + trade_data.get('status'), + trade_data.get('error_message'), + json.dumps(trade_data.get('raw_data', {}), ensure_ascii=False) + )) + conn.commit() + except Exception as e: + print(f"写入数据库失败: {e}") + + def check_existing_trade(self, market_slug: str, status: str = 'success') -> bool: + """ + 检查指定市场是否已有成功交易记录 + + Args: + market_slug: 市场slug + status: 交易状态 (默认 'success', 也可以是 'test') + + Returns: + True 如果已有成功交易, 否则 False + """ + try: + with self.get_connection() as conn: + cursor = conn.cursor() + cursor.execute(''' + SELECT COUNT(*) FROM trades + WHERE market_slug = ? AND status = ? + ''', (market_slug, status)) + count = cursor.fetchone()[0] + return count > 0 + except Exception as e: + print(f"检查已存在交易失败: {e}") + return False