diff --git a/.github/workflows/cicd.yaml b/.github/workflows/cicd.yaml index 9dcf757..7130a7b 100644 --- a/.github/workflows/cicd.yaml +++ b/.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 diff --git a/backend/config/config.yaml b/backend/config/config.yaml index 179610e..7fe7490 100644 --- a/backend/config/config.yaml +++ b/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" diff --git a/data/block_chain/collect_binance.py b/data/block_chain/collect_binance.py index 7781827..ce66cfe 100644 --- a/data/block_chain/collect_binance.py +++ b/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) diff --git a/data/block_chain/collect_uniswap.py b/data/block_chain/collect_uniswap.py index 941d1c3..e940ff2 100644 --- a/data/block_chain/collect_uniswap.py +++ b/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 diff --git a/data/block_chain/task.py b/data/block_chain/task.py index cdd3320..e18d4e6 100644 --- a/data/block_chain/task.py +++ b/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 diff --git a/data/config/config.yaml b/data/config/config.yaml index f5e7d30..9a1dc2d 100644 --- a/data/config/config.yaml +++ b/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 diff --git a/data/protos/task_pb2.py b/data/protos/task_pb2.py index b904e0a..3eb0009 100644 --- a/data/protos/task_pb2.py +++ b/data/protos/task_pb2.py @@ -1,7 +1,7 @@ # -*- coding: utf-8 -*- # Generated by the protocol buffer compiler. DO NOT EDIT! # NO CHECKED-IN PROTOBUF GENCODE -# source: task.proto +# source: protos/task.proto # Protobuf Python Version: 6.31.1 """Generated protocol buffer code.""" from google.protobuf import descriptor as _descriptor @@ -11,7 +11,7 @@ from google.protobuf.internal import builder as _builder _runtime_version.ValidateProtobufRuntimeVersion( - _runtime_version.Domain.PUBLIC, 6, 31, 1, "", "task.proto" + _runtime_version.Domain.PUBLIC, 6, 31, 1, "", "protos/task.proto" ) # @@protoc_insertion_point(imports) @@ -19,31 +19,33 @@ DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( - b'\n\ntask.proto\x12\x07task.v1"W\n\x15\x43ollectBinanceRequest\x12\x0f\n\x07task_id\x18\x01 \x01(\t\x12\x19\n\x11import_percentage\x18\x02 \x01(\x05\x12\x12\n\nchunk_size\x18\x03 \x01(\x05"`\n\x15\x43ollectUniswapRequest\x12\x0f\n\x07task_id\x18\x01 \x01(\t\x12\x14\n\x0cpool_address\x18\x02 \x01(\t\x12\x10\n\x08start_ts\x18\x03 \x01(\x05\x12\x0e\n\x06\x65nd_ts\x18\x04 \x01(\x05"\xf8\x01\n\x14ProcessPricesRequest\x12\x0f\n\x07task_id\x18\x01 \x01(\t\x12\x12\n\nstart_date\x18\x02 \x01(\x05\x12\x10\n\x08\x65nd_date\x18\x03 \x01(\x05\x12\x1c\n\x14\x61ggregation_interval\x18\x04 \x01(\t\x12\x11\n\toverwrite\x18\x05 \x01(\x08\x12\x44\n\x0c\x64\x62_overrides\x18\x06 \x03(\x0b\x32..task.v1.ProcessPricesRequest.DbOverridesEntry\x1a\x32\n\x10\x44\x62OverridesEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01"]\n\x0e\x41nalyseRequest\x12\x0f\n\x07task_id\x18\x01 \x01(\t\x12\x10\n\x08\x62\x61tch_id\x18\x02 \x01(\x05\x12\x11\n\toverwrite\x18\x03 \x01(\x08\x12\x15\n\rstrategy_json\x18\x04 \x01(\t"D\n\x0cTaskResponse\x12\x0f\n\x07task_id\x18\x01 \x01(\t\x12#\n\x06status\x18\x02 \x01(\x0e\x32\x13.task.v1.TaskStatus*p\n\nTaskStatus\x12\x17\n\x13TASK_STATUS_RUNNING\x10\x00\x12\x17\n\x13TASK_STATUS_SUCCESS\x10\x01\x12\x16\n\x12TASK_STATUS_FAILED\x10\x02\x12\x18\n\x14TASK_STATUS_CANCELED\x10\x03\x32\xa1\x02\n\x0bTaskService\x12G\n\x0e\x43ollectBinance\x12\x1e.task.v1.CollectBinanceRequest\x1a\x15.task.v1.TaskResponse\x12G\n\x0e\x43ollectUniswap\x12\x1e.task.v1.CollectUniswapRequest\x1a\x15.task.v1.TaskResponse\x12\x45\n\rProcessPrices\x12\x1d.task.v1.ProcessPricesRequest\x1a\x15.task.v1.TaskResponse\x12\x39\n\x07\x41nalyse\x12\x17.task.v1.AnalyseRequest\x1a\x15.task.v1.TaskResponseB\x1bZ\x19\x62\x61\x63kend/pkg/taskpb;taskpbb\x06proto3' + b'\n\x11protos/task.proto\x12\x07task.v1"W\n\x15\x43ollectBinanceRequest\x12\x0f\n\x07task_id\x18\x01 \x01(\t\x12\x19\n\x11import_percentage\x18\x02 \x01(\x05\x12\x12\n\nchunk_size\x18\x03 \x01(\x05"O\n\x1a\x43ollectBinaceByDateRequest\x12\x0f\n\x07task_id\x18\x01 \x01(\t\x12\x10\n\x08start_ts\x18\x02 \x01(\x05\x12\x0e\n\x06\x65nd_ts\x18\x03 \x01(\x05"`\n\x15\x43ollectUniswapRequest\x12\x0f\n\x07task_id\x18\x01 \x01(\t\x12\x14\n\x0cpool_address\x18\x02 \x01(\t\x12\x10\n\x08start_ts\x18\x03 \x01(\x05\x12\x0e\n\x06\x65nd_ts\x18\x04 \x01(\x05"\xf8\x01\n\x14ProcessPricesRequest\x12\x0f\n\x07task_id\x18\x01 \x01(\t\x12\x12\n\nstart_date\x18\x02 \x01(\x05\x12\x10\n\x08\x65nd_date\x18\x03 \x01(\x05\x12\x1c\n\x14\x61ggregation_interval\x18\x04 \x01(\t\x12\x11\n\toverwrite\x18\x05 \x01(\x08\x12\x44\n\x0c\x64\x62_overrides\x18\x06 \x03(\x0b\x32..task.v1.ProcessPricesRequest.DbOverridesEntry\x1a\x32\n\x10\x44\x62OverridesEntry\x12\x0b\n\x03key\x18\x01 \x01(\t\x12\r\n\x05value\x18\x02 \x01(\t:\x02\x38\x01"]\n\x0e\x41nalyseRequest\x12\x0f\n\x07task_id\x18\x01 \x01(\t\x12\x10\n\x08\x62\x61tch_id\x18\x02 \x01(\x05\x12\x11\n\toverwrite\x18\x03 \x01(\x08\x12\x15\n\rstrategy_json\x18\x04 \x01(\t"D\n\x0cTaskResponse\x12\x0f\n\x07task_id\x18\x01 \x01(\t\x12#\n\x06status\x18\x02 \x01(\x0e\x32\x13.task.v1.TaskStatus*p\n\nTaskStatus\x12\x17\n\x13TASK_STATUS_RUNNING\x10\x00\x12\x17\n\x13TASK_STATUS_SUCCESS\x10\x01\x12\x16\n\x12TASK_STATUS_FAILED\x10\x02\x12\x18\n\x14TASK_STATUS_CANCELED\x10\x03\x32\xf5\x02\n\x0bTaskService\x12G\n\x0e\x43ollectBinance\x12\x1e.task.v1.CollectBinanceRequest\x1a\x15.task.v1.TaskResponse\x12R\n\x14\x43ollectBinanceByDate\x12#.task.v1.CollectBinaceByDateRequest\x1a\x15.task.v1.TaskResponse\x12G\n\x0e\x43ollectUniswap\x12\x1e.task.v1.CollectUniswapRequest\x1a\x15.task.v1.TaskResponse\x12\x45\n\rProcessPrices\x12\x1d.task.v1.ProcessPricesRequest\x1a\x15.task.v1.TaskResponse\x12\x39\n\x07\x41nalyse\x12\x17.task.v1.AnalyseRequest\x1a\x15.task.v1.TaskResponseB\x1bZ\x19\x62\x61\x63kend/pkg/taskpb;taskpbb\x06proto3' ) _globals = globals() _builder.BuildMessageAndEnumDescriptors(DESCRIPTOR, _globals) -_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, "task_pb2", _globals) +_builder.BuildTopDescriptorsAndMessages(DESCRIPTOR, "protos.task_pb2", _globals) if not _descriptor._USE_C_DESCRIPTORS: _globals["DESCRIPTOR"]._loaded_options = None _globals["DESCRIPTOR"]._serialized_options = b"Z\031backend/pkg/taskpb;taskpb" _globals["_PROCESSPRICESREQUEST_DBOVERRIDESENTRY"]._loaded_options = None _globals["_PROCESSPRICESREQUEST_DBOVERRIDESENTRY"]._serialized_options = b"8\001" - _globals["_TASKSTATUS"]._serialized_start = 626 - _globals["_TASKSTATUS"]._serialized_end = 738 - _globals["_COLLECTBINANCEREQUEST"]._serialized_start = 23 - _globals["_COLLECTBINANCEREQUEST"]._serialized_end = 110 - _globals["_COLLECTUNISWAPREQUEST"]._serialized_start = 112 - _globals["_COLLECTUNISWAPREQUEST"]._serialized_end = 208 - _globals["_PROCESSPRICESREQUEST"]._serialized_start = 211 - _globals["_PROCESSPRICESREQUEST"]._serialized_end = 459 - _globals["_PROCESSPRICESREQUEST_DBOVERRIDESENTRY"]._serialized_start = 409 - _globals["_PROCESSPRICESREQUEST_DBOVERRIDESENTRY"]._serialized_end = 459 - _globals["_ANALYSEREQUEST"]._serialized_start = 461 - _globals["_ANALYSEREQUEST"]._serialized_end = 554 - _globals["_TASKRESPONSE"]._serialized_start = 556 - _globals["_TASKRESPONSE"]._serialized_end = 624 - _globals["_TASKSERVICE"]._serialized_start = 741 - _globals["_TASKSERVICE"]._serialized_end = 1030 + _globals["_TASKSTATUS"]._serialized_start = 714 + _globals["_TASKSTATUS"]._serialized_end = 826 + _globals["_COLLECTBINANCEREQUEST"]._serialized_start = 30 + _globals["_COLLECTBINANCEREQUEST"]._serialized_end = 117 + _globals["_COLLECTBINACEBYDATEREQUEST"]._serialized_start = 119 + _globals["_COLLECTBINACEBYDATEREQUEST"]._serialized_end = 198 + _globals["_COLLECTUNISWAPREQUEST"]._serialized_start = 200 + _globals["_COLLECTUNISWAPREQUEST"]._serialized_end = 296 + _globals["_PROCESSPRICESREQUEST"]._serialized_start = 299 + _globals["_PROCESSPRICESREQUEST"]._serialized_end = 547 + _globals["_PROCESSPRICESREQUEST_DBOVERRIDESENTRY"]._serialized_start = 497 + _globals["_PROCESSPRICESREQUEST_DBOVERRIDESENTRY"]._serialized_end = 547 + _globals["_ANALYSEREQUEST"]._serialized_start = 549 + _globals["_ANALYSEREQUEST"]._serialized_end = 642 + _globals["_TASKRESPONSE"]._serialized_start = 644 + _globals["_TASKRESPONSE"]._serialized_end = 712 + _globals["_TASKSERVICE"]._serialized_start = 829 + _globals["_TASKSERVICE"]._serialized_end = 1202 # @@protoc_insertion_point(module_scope) diff --git a/data/protos/task_pb2_grpc.py b/data/protos/task_pb2_grpc.py index d5202bf..b32ca65 100644 --- a/data/protos/task_pb2_grpc.py +++ b/data/protos/task_pb2_grpc.py @@ -4,7 +4,7 @@ import grpc -from . import task_pb2 as task__pb2 +from protos import task_pb2 as protos_dot_task__pb2 GRPC_GENERATED_VERSION = "1.76.0" GRPC_VERSION = grpc.__version__ @@ -22,7 +22,7 @@ if _version_not_supported: raise RuntimeError( f"The grpc package installed is at version {GRPC_VERSION}," - + " but the generated code in task_pb2_grpc.py depends on" + + " but the generated code in protos/task_pb2_grpc.py depends on" + f" grpcio>={GRPC_GENERATED_VERSION}." + f" Please upgrade your grpc module to grpcio>={GRPC_GENERATED_VERSION}" + f" or downgrade your generated code using grpcio-tools<={GRPC_VERSION}." @@ -40,26 +40,32 @@ def __init__(self, channel): """ self.CollectBinance = channel.unary_unary( "/task.v1.TaskService/CollectBinance", - request_serializer=task__pb2.CollectBinanceRequest.SerializeToString, - response_deserializer=task__pb2.TaskResponse.FromString, + request_serializer=protos_dot_task__pb2.CollectBinanceRequest.SerializeToString, + response_deserializer=protos_dot_task__pb2.TaskResponse.FromString, + _registered_method=True, + ) + self.CollectBinanceByDate = channel.unary_unary( + "/task.v1.TaskService/CollectBinanceByDate", + request_serializer=protos_dot_task__pb2.CollectBinaceByDateRequest.SerializeToString, + response_deserializer=protos_dot_task__pb2.TaskResponse.FromString, _registered_method=True, ) self.CollectUniswap = channel.unary_unary( "/task.v1.TaskService/CollectUniswap", - request_serializer=task__pb2.CollectUniswapRequest.SerializeToString, - response_deserializer=task__pb2.TaskResponse.FromString, + request_serializer=protos_dot_task__pb2.CollectUniswapRequest.SerializeToString, + response_deserializer=protos_dot_task__pb2.TaskResponse.FromString, _registered_method=True, ) self.ProcessPrices = channel.unary_unary( "/task.v1.TaskService/ProcessPrices", - request_serializer=task__pb2.ProcessPricesRequest.SerializeToString, - response_deserializer=task__pb2.TaskResponse.FromString, + request_serializer=protos_dot_task__pb2.ProcessPricesRequest.SerializeToString, + response_deserializer=protos_dot_task__pb2.TaskResponse.FromString, _registered_method=True, ) self.Analyse = channel.unary_unary( "/task.v1.TaskService/Analyse", - request_serializer=task__pb2.AnalyseRequest.SerializeToString, - response_deserializer=task__pb2.TaskResponse.FromString, + request_serializer=protos_dot_task__pb2.AnalyseRequest.SerializeToString, + response_deserializer=protos_dot_task__pb2.TaskResponse.FromString, _registered_method=True, ) @@ -73,6 +79,12 @@ def CollectBinance(self, request, context): context.set_details("Method not implemented!") raise NotImplementedError("Method not implemented!") + def CollectBinanceByDate(self, request, context): + """按日期收集币安数据""" + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + context.set_details("Method not implemented!") + raise NotImplementedError("Method not implemented!") + def CollectUniswap(self, request, context): """收集Uniswap数据""" context.set_code(grpc.StatusCode.UNIMPLEMENTED) @@ -96,23 +108,28 @@ def add_TaskServiceServicer_to_server(servicer, server): rpc_method_handlers = { "CollectBinance": grpc.unary_unary_rpc_method_handler( servicer.CollectBinance, - request_deserializer=task__pb2.CollectBinanceRequest.FromString, - response_serializer=task__pb2.TaskResponse.SerializeToString, + request_deserializer=protos_dot_task__pb2.CollectBinanceRequest.FromString, + response_serializer=protos_dot_task__pb2.TaskResponse.SerializeToString, + ), + "CollectBinanceByDate": grpc.unary_unary_rpc_method_handler( + servicer.CollectBinanceByDate, + request_deserializer=protos_dot_task__pb2.CollectBinaceByDateRequest.FromString, + response_serializer=protos_dot_task__pb2.TaskResponse.SerializeToString, ), "CollectUniswap": grpc.unary_unary_rpc_method_handler( servicer.CollectUniswap, - request_deserializer=task__pb2.CollectUniswapRequest.FromString, - response_serializer=task__pb2.TaskResponse.SerializeToString, + request_deserializer=protos_dot_task__pb2.CollectUniswapRequest.FromString, + response_serializer=protos_dot_task__pb2.TaskResponse.SerializeToString, ), "ProcessPrices": grpc.unary_unary_rpc_method_handler( servicer.ProcessPrices, - request_deserializer=task__pb2.ProcessPricesRequest.FromString, - response_serializer=task__pb2.TaskResponse.SerializeToString, + request_deserializer=protos_dot_task__pb2.ProcessPricesRequest.FromString, + response_serializer=protos_dot_task__pb2.TaskResponse.SerializeToString, ), "Analyse": grpc.unary_unary_rpc_method_handler( servicer.Analyse, - request_deserializer=task__pb2.AnalyseRequest.FromString, - response_serializer=task__pb2.TaskResponse.SerializeToString, + request_deserializer=protos_dot_task__pb2.AnalyseRequest.FromString, + response_serializer=protos_dot_task__pb2.TaskResponse.SerializeToString, ), } generic_handler = grpc.method_handlers_generic_handler( @@ -143,8 +160,38 @@ def CollectBinance( request, target, "/task.v1.TaskService/CollectBinance", - task__pb2.CollectBinanceRequest.SerializeToString, - task__pb2.TaskResponse.FromString, + protos_dot_task__pb2.CollectBinanceRequest.SerializeToString, + protos_dot_task__pb2.TaskResponse.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True, + ) + + @staticmethod + def CollectBinanceByDate( + request, + target, + options=(), + channel_credentials=None, + call_credentials=None, + insecure=False, + compression=None, + wait_for_ready=None, + timeout=None, + metadata=None, + ): + return grpc.experimental.unary_unary( + request, + target, + "/task.v1.TaskService/CollectBinanceByDate", + protos_dot_task__pb2.CollectBinaceByDateRequest.SerializeToString, + protos_dot_task__pb2.TaskResponse.FromString, options, channel_credentials, insecure, @@ -173,8 +220,8 @@ def CollectUniswap( request, target, "/task.v1.TaskService/CollectUniswap", - task__pb2.CollectUniswapRequest.SerializeToString, - task__pb2.TaskResponse.FromString, + protos_dot_task__pb2.CollectUniswapRequest.SerializeToString, + protos_dot_task__pb2.TaskResponse.FromString, options, channel_credentials, insecure, @@ -203,8 +250,8 @@ def ProcessPrices( request, target, "/task.v1.TaskService/ProcessPrices", - task__pb2.ProcessPricesRequest.SerializeToString, - task__pb2.TaskResponse.FromString, + protos_dot_task__pb2.ProcessPricesRequest.SerializeToString, + protos_dot_task__pb2.TaskResponse.FromString, options, channel_credentials, insecure, @@ -233,8 +280,8 @@ def Analyse( request, target, "/task.v1.TaskService/Analyse", - task__pb2.AnalyseRequest.SerializeToString, - task__pb2.TaskResponse.FromString, + protos_dot_task__pb2.AnalyseRequest.SerializeToString, + protos_dot_task__pb2.TaskResponse.FromString, options, channel_credentials, insecure, diff --git a/data/server.py b/data/server.py index 1d629f2..86a2660 100644 --- a/data/server.py +++ b/data/server.py @@ -152,6 +152,53 @@ def run_task(): status=TaskStatus.TASK_STATUS_RUNNING, ) + def CollectBinanceByDate(self, request, context): + """ + 按日期收集币安数据 + 在后台线程中执行任务，立即返回运行状态 + """ + task_id = request.task_id + start_ts = request.start_ts + end_ts = request.end_ts + + logger.info( + f"收到按日期收集币安数据请求: task_id={task_id}, " + f"start_ts={start_ts}, end_ts={end_ts}" + ) + + def run_task(): + """在后台线程中执行任务""" + try: + logger.info(f"开始执行任务 {task_id}: 按日期收集币安数据") + mark_task_started(task_id) + log_task_event(task_id, "INFO", "开始按日期收集 Binance 数据") + collect_binance.collect_binance_by_date( + task_id=task_id, + start_ts=start_ts, + end_ts=end_ts, + ) + logger.info(f"任务 {task_id} 执行成功: 按日期收集币安数据") + log_task_event(task_id, "INFO", "按日期收集 Binance 数据完成") + mark_task_finished( + task_id, + TaskStatus.TASK_STATUS_SUCCESS, + "Binance 数据按日期收集完成", + ) + except Exception as e: + logger.error(f"任务 {task_id} 执行失败: {e}") + log_task_event(task_id, "ERROR", f"按日期收集 Binance 数据失败: {e}") + mark_task_finished(task_id, TaskStatus.TASK_STATUS_FAILED, str(e)) + + # 在后台线程中启动任务 + thread = threading.Thread(target=run_task, daemon=True) + thread.start() + + # 立即返回运行状态 + return TaskResponse( + task_id=task_id, + status=TaskStatus.TASK_STATUS_RUNNING, + ) + def Analyse(self, request, context): """ 执行套利分析任务 diff --git a/data/test/test_collect_binance.py b/data/test/test_collect_binance.py index 37d7b73..f12ac5c 100644 --- a/data/test/test_collect_binance.py +++ b/data/test/test_collect_binance.py @@ -233,7 +233,7 @@ def test_process_chunk_handles_db_error(self, sample_chunk, mock_db_connection): ) # 验证任务状态被更新为失败 mock_update_status.assert_called_once_with( - "test_task", "TASK_STATUS_FAILED" + "test_task", "FAILED" ) def test_process_chunk_stops_at_target_rows(self, mock_db_connection): @@ -407,7 +407,7 @@ def side_effect(task_id, chunk, idx, counter, target): mock_process_chunk.side_effect = side_effect - rows_counter = import_data_to_database("test_task", None, None, 100) + rows_counter = import_data_to_database("test_task", "test.csv", None, None, 100) assert len(rows_counter) == 2 assert rows_counter[0] == 2 # 处理的行数 @@ -454,7 +454,7 @@ def side_effect(task_id, chunk, idx, counter, target): mock_process_chunk.side_effect = side_effect - rows_counter = import_data_to_database("test_task", 3, None, 100) + rows_counter = import_data_to_database("test_task", "test.csv", 3, None, 100) assert len(rows_counter) == 2 # 应该达到目标行数并停止 @@ -473,9 +473,9 @@ def test_import_data_to_database_file_not_found( mock_read_csv.side_effect = FileNotFoundError("File not found") with pytest.raises(FileNotFoundError): - import_data_to_database("test_task", None, None, 100) + import_data_to_database("test_task", "test.csv", None, None, 100) - mock_update_status.assert_called_once_with("test_task", "TASK_STATUS_FAILED") + mock_update_status.assert_called_once_with("test_task", "FAILED") @patch("block_chain.collect_binance.pd.read_csv") @patch("block_chain.collect_binance.update_task_status") @@ -490,9 +490,9 @@ def test_import_data_to_database_general_exception( mock_read_csv.side_effect = Exception("Unexpected error") with pytest.raises(Exception, match="Unexpected error"): - import_data_to_database("test_task", None, None, 100) + import_data_to_database("test_task", "test.csv", None, None, 100) - mock_update_status.assert_called_once_with("test_task", "TASK_STATUS_FAILED") + mock_update_status.assert_called_once_with("test_task", "FAILED") @patch("block_chain.collect_binance.pd.read_csv") @patch("block_chain.collect_binance.check_task", return_value=False) @@ -544,7 +544,7 @@ def side_effect(task_id, chunk, idx, counter, target): mock_process_chunk.side_effect = side_effect - rows_counter = import_data_to_database("test_task", None, None, 100) + rows_counter = import_data_to_database("test_task", "test.csv", None, None, 100) assert len(rows_counter) == 2 assert rows_counter[0] == 4 # 处理了4行 @@ -604,7 +604,7 @@ def side_effect(task_id, chunk, idx, counter, target): mock_process_chunk.side_effect = side_effect - rows_counter = import_data_to_database("test_task", 3, None, 100) + rows_counter = import_data_to_database("test_task", "test.csv", 3, None, 100) assert len(rows_counter) == 2 # 应该只处理第一个chunk就停止（因为达到目标行数3） diff --git a/data/test/test_task.py b/data/test/test_task.py index adfd814..14dde67 100644 --- a/data/test/test_task.py +++ b/data/test/test_task.py @@ -58,7 +58,7 @@ def test_check_task_canceled(self, mock_connect): mock_conn.cursor.return_value.__exit__ = lambda *args: None # Mock返回已取消状态 - mock_cursor.fetchone.return_value = ("TASK_STATUS_CANCELED",) + mock_cursor.fetchone.return_value = ("CANCELLED",) result = check_task("test_task_1") diff --git a/protos/task.proto b/protos/task.proto index db81199..65a6431 100644 --- a/protos/task.proto +++ b/protos/task.proto @@ -7,6 +7,8 @@ option go_package = "backend/pkg/taskpb;taskpb"; service TaskService { // 收集币安数据 rpc CollectBinance(CollectBinanceRequest) returns (TaskResponse); + // 按日期收集币安数据 + rpc CollectBinanceByDate(CollectBinaceByDateRequest) returns (TaskResponse); // 收集Uniswap数据 rpc CollectUniswap(CollectUniswapRequest) returns (TaskResponse); // 处理价格数据 @@ -29,6 +31,12 @@ message CollectBinanceRequest { int32 chunk_size = 3; // 分块大小 } +message CollectBinaceByDateRequest { + string task_id = 1; // 任务的唯一标识符 + int32 start_ts = 2; // 起始时间 + int32 end_ts = 3; // 终止时间 +} + message CollectUniswapRequest { string task_id = 1; // 任务的唯一标识符 string pool_address = 2; // 池地址