add：添加按日期导入binance数据

.github/workflows/cicd.yaml

@@ -16,15 +16,15 @@ jobs:
       with:
         go-version: '1.24'
 
-    #- name: Install and configure PostgreSQL
-    #  run: |
-    #    docker pull postgres
-    #    docker run -d \
-    #    --name postgresql \
-    #    -p 5432:5432 \
-    #    -e POSTGRES_PASSWORD="123456" \
-    #    postgres
-    #    docker ps
+    - name: Install and configure PostgreSQL
+      run: |
+        docker pull postgres
+        docker run -d \
+        --name postgresql \
+        -p 5432:5432 \
+        -e POSTGRES_PASSWORD="123456" \
+        postgres
+        docker ps
 
     - name: Download Go dependencies
       working-directory: ./backend

backend/config/config.yaml

@@ -1,8 +1,8 @@
 db:
-  host: 121.196.205.18
+  host: localhost
   port: 5432
   username: postgres
-  password: etrade2025
+  password: 123456
   database: etrade
 grpc:
   port: ":50060"

data/block_chain/collect_binance.py

@@ -1,13 +1,18 @@
 import argparse
 import io
 import math
+import os
 import sys
+import tempfile
 import time
 import traceback
+import zipfile
+from datetime import datetime, timedelta, timezone
 from typing import Optional
 
 import pandas as pd
 import psycopg2
+import requests
 import yaml
 from loguru import logger
 
@@ -18,7 +23,6 @@
     config = yaml.safe_load(file)
 
 db_config = config.get("db", {})
-csv_path = "./ETHUSDT-trades-2025-09.csv"
 COLUMN_NAMES = ["id", "price", "qty", "quoteQty", "time", "isBuyerMaker", "isBestMatch"]
 DTYPE_MAP = {
     "id": "int64",
@@ -56,11 +60,11 @@ def count_lines(task_id: str, filepath: str) -> Optional[int]:
         return count
     except FileNotFoundError:
         logger.error(f"找不到CSV文件 '{filepath}'。请检查路径是否正确。")
-        update_task_status(task_id, "TASK_STATUS_FAILED")
+        update_task_status(task_id, "FAILED")
         raise
     except Exception as e:
         logger.warning(f"估算行数失败: {e}. 无法按百分比导入。")
-        update_task_status(task_id, "TASK_STATUS_FAILED")
+        update_task_status(task_id, "FAILED")
         raise
 
 
@@ -118,12 +122,13 @@ def process_chunk(
         return True, original_chunk_len, rows_imported, should_stop
     except Exception as e:
         logger.error(f"处理分块时发生意外错误: {e}")
-        update_task_status(task_id, "TASK_STATUS_FAILED")
+        update_task_status(task_id, "FAILED")
         raise
 
 
 def import_data_to_database(
     task_id: str,
+    csv_path: str,
     target_rows: Optional[int],
     total_lines: Optional[int],
     chunk_size: int,
@@ -168,12 +173,12 @@ def import_data_to_database(
                 break
     except FileNotFoundError:
         logger.error(f"找不到CSV文件 '{csv_path}'。请检查路径是否正确。")
-        update_task_status(task_id, "TASK_STATUS_FAILED")
+        update_task_status(task_id, "FAILED")
         raise
     except Exception as e:
         logger.error(f"处理文件时发生意外错误: {e}")
         traceback.print_exc(file=sys.stderr)
-        update_task_status(task_id, "TASK_STATUS_FAILED")
+        update_task_status(task_id, "FAILED")
         raise
     return rows_counter
 
@@ -199,20 +204,175 @@ def collect_binance(
             return 0
         target_rows = _calc_target_rows(total_lines, import_percentage)
         rows_counter = import_data_to_database(
-            task_id, target_rows, total_lines, chunk_size
+            task_id, csv_path, target_rows, total_lines, chunk_size
         )
         total_time = time.time() - start_time
         if check_task(task_id):
             logger.info(f"任务 {task_id} 已取消，停止导入 Binance 数据")
             return 0
         logger.info(f"成功导入 {rows_counter[1]} 行，耗时 {total_time:.2f}s")
-        update_task_status(task_id, "TASK_STATUS_SUCCESS")
+        update_task_status(task_id, "SUCCESS")
         return rows_counter[1]
     except Exception as e:
         logger.error(f"导入 Binance 数据失败: {e}")
-        update_task_status(task_id, "TASK_STATUS_FAILED")
+        update_task_status(task_id, "FAILED")
+        raise
+
+
+def download_binance_file(
+    task_id: str, date_str: str, symbol: str = "ETHUSDT"
+) -> Optional[str]:
+    """
+    描述：从币安数据源下载指定日期的交易数据文件
+    参数：task_id: 任务ID, date_str: 日期字符串 (YYYY-MM-DD), symbol: 交易对符号
+    返回值：下载的CSV文件路径，如果失败返回None
+    """
+    base_url = "https://data.binance.vision/data/spot/daily/trades"
+    url = f"{base_url}/{symbol}/{symbol}-trades-{date_str}.zip"
+
+    try:
+        logger.info(f"正在下载币安数据: {url}")
+        response = requests.get(url, stream=True, timeout=30)
+        response.raise_for_status()
+
+        # 创建临时目录保存文件
+        temp_dir = tempfile.mkdtemp(prefix="binance_")
+        zip_path = os.path.join(temp_dir, f"{symbol}-trades-{date_str}.zip")
+
+        # 下载zip文件
+        with open(zip_path, "wb") as f:
+            for chunk in response.iter_content(chunk_size=8192):
+                if chunk:
+                    f.write(chunk)
+                    if check_task(task_id):
+                        logger.info(f"任务 {task_id} 已取消，停止下载")
+                        os.remove(zip_path)
+                        os.rmdir(temp_dir)
+                        return None
+
+        # 解压文件
+        csv_path = None
+        with zipfile.ZipFile(zip_path, "r") as zip_ref:
+            # 获取zip文件中的CSV文件名
+            csv_files = [f for f in zip_ref.namelist() if f.endswith(".csv")]
+            if not csv_files:
+                logger.error(f"ZIP文件中没有找到CSV文件: {zip_path}")
+                os.remove(zip_path)
+                os.rmdir(temp_dir)
+                return None
+
+            # 解压第一个CSV文件
+            csv_filename = csv_files[0]
+            zip_ref.extract(csv_filename, temp_dir)
+            csv_path = os.path.join(temp_dir, csv_filename)
+
+        # 删除zip文件
+        os.remove(zip_path)
+        logger.info(f"成功下载并解压: {csv_path}")
+        return csv_path
+
+    except requests.exceptions.RequestException as e:
+        logger.error(f"下载币安数据失败: {e}")
+        return None
+    except zipfile.BadZipFile as e:
+        logger.error(f"解压文件失败: {e}")
+        return None
+    except Exception as e:
+        logger.error(f"处理文件时发生错误: {e}")
+        return None
+
+
+def collect_binance_by_date(
+    task_id: str,
+    start_ts: int,
+    end_ts: int,
+    symbol: str = "ETHUSDT",
+    chunk_size: int = 1000000,
+) -> int:
+    """
+    描述：按日期范围收集币安数据
+    参数：
+        task_id: 任务ID
+        start_ts: 起始时间戳（秒级）
+        end_ts: 终止时间戳（秒级）
+        symbol: 交易对符号，默认为ETHUSDT
+        chunk_size: 分块大小，默认1000000
+    返回值：导入的总行数
+    """
+    try:
+        start_time = time.time()
+
+        # 将时间戳转换为日期
+        start_date = datetime.fromtimestamp(start_ts, tz=timezone.utc)
+        end_date = datetime.fromtimestamp(end_ts, tz=timezone.utc)
+
+        logger.info(f"开始按日期收集币安数据: {start_date.date()} 到 {end_date.date()}")
+
+        total_rows_imported = 0
+        temp_files = []  # 记录临时文件，用于清理
+
+        # 遍历日期范围
+        current_date = start_date.date()
+        end_date_only = end_date.date()
+
+        while current_date <= end_date_only:
+            if check_task(task_id):
+                logger.info(f"任务 {task_id} 已取消，停止收集 Binance 数据")
+                break
+
+            date_str = current_date.strftime("%Y-%m-%d")
+            logger.info(f"正在处理日期: {date_str}")
+
+            # 下载文件
+            csv_path = download_binance_file(task_id, date_str, symbol)
+
+            if csv_path is None:
+                logger.warning(f"跳过日期 {date_str}，下载失败")
+                current_date += timedelta(days=1)
+                continue
+
+            temp_files.append(csv_path)
+            temp_files.append(os.path.dirname(csv_path))  # 临时目录
+
+            try:
+                # 导入数据（导入全部数据，不限制百分比）
+                rows_counter = import_data_to_database(
+                    task_id, csv_path, None, None, chunk_size
+                )
+                total_rows_imported += rows_counter[1]
+                logger.info(f"日期 {date_str} 导入完成，导入 {rows_counter[1]} 行")
+            except Exception as e:
+                logger.error(f"导入日期 {date_str} 的数据失败: {e}")
+                # 继续处理下一个日期，不中断整个任务
+
+            # 清理临时文件
+            try:
+                if os.path.exists(csv_path):
+                    os.remove(csv_path)
+                temp_dir = os.path.dirname(csv_path)
+                if os.path.exists(temp_dir) and os.path.isdir(temp_dir):
+                    os.rmdir(temp_dir)
+            except Exception as e:
+                logger.warning(f"清理临时文件失败: {e}")
+
+            current_date += timedelta(days=1)
+
+        total_time = time.time() - start_time
+
+        if check_task(task_id):
+            logger.info(f"任务 {task_id} 已取消，停止收集 Binance 数据")
+            return 0
+
+        logger.info(f"成功导入 {total_rows_imported} 行，耗时 {total_time:.2f}s")
+        update_task_status(task_id, "SUCCESS")
+        return total_rows_imported
+
+    except Exception as e:
+        logger.error(f"按日期收集 Binance 数据失败: {e}")
+        traceback.print_exc(file=sys.stderr)
+        update_task_status(task_id, "FAILED")
         raise
 
 
 if __name__ == "__main__":
-    collect_binance("1", csv_path, 1, 1000000)
+    collect_binance("1", "./ETHUSDT-trades-2025-12-23.csv", 1, 1000000)

data/block_chain/collect_uniswap.py

@@ -151,11 +151,11 @@ def collect_uniswap(task_id: str, pool_address: str, start_ts: int, end_ts: int)
             logger.info(f"任务 {task_id} 已取消，停止写入 Uniswap 数据")
             return 0
         rows_counter = process_and_store_uniswap_data(task_id, swaps)
-        update_task_status(task_id, "TASK_STATUS_SUCCESS")
+        update_task_status(task_id, "SUCCESS")
         return rows_counter
     except Exception as e:
         logger.error(f"获取Uniswap数据失败: {e}")
-        update_task_status(task_id, "TASK_STATUS_FAILED")
+        update_task_status(task_id, "FAILED")
         return 0
 
 

data/block_chain/task.py

@@ -31,7 +31,7 @@ def check_task(task_id: str):
                 if result is None:
                     logger.error(f"任务 {task_id} 不存在")
                     return True
-                return result[0] == "TASK_STATUS_CANCELED"
+                return result[0] == "CANCELLED"
     except Exception as e:
         logger.error(f"检查任务 {task_id} 失败: {e}")
         return True

data/config/config.yaml

@@ -1,9 +1,9 @@
 db:
-  host: 121.196.205.18
+  host: localhost
   port: 5432
   database: etrade
   username: postgres
-  password: etrade2025
+  password: 123456
 
 the_graph:
   api_key: 9f9faba5da813868926b3337fb728af5