From a3cc210bed2e386030f54bc616f04a637ec8a9e3 Mon Sep 17 00:00:00 2001 From: cagoya <2561694274@qq.com> Date: Wed, 24 Dec 2025 23:36:24 +0800 Subject: [PATCH 1/4] =?UTF-8?q?add=EF=BC=9A=E6=B7=BB=E5=8A=A0=E6=8C=89?= =?UTF-8?q?=E6=97=A5=E6=9C=9F=E5=AF=BC=E5=85=A5binance=E6=95=B0=E6=8D=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/config/config.yaml | 4 +- data/block_chain/collect_binance.py | 164 ++++++++++++++- data/config/config.yaml | 4 +- data/protos/task_pb2.py | 60 +++--- data/protos/task_pb2_grpc.py | 310 +++++++++++++++------------- data/server.py | 45 ++++ protos/task.proto | 8 + 7 files changed, 421 insertions(+), 174 deletions(-) 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..9cd057d 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", @@ -124,6 +128,7 @@ def process_chunk( def import_data_to_database( task_id: str, + csv_path: str, target_rows: Optional[int], total_lines: Optional[int], chunk_size: int, @@ -199,7 +204,7 @@ 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): @@ -214,5 +219,158 @@ def collect_binance( 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, "TASK_STATUS_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, "TASK_STATUS_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/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..6a94914 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 @@ -9,41 +9,47 @@ from google.protobuf import runtime_version as _runtime_version from google.protobuf import symbol_database as _symbol_database 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) _sym_db = _symbol_database.Default() -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' -) + + +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(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['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=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..29268ce 100644 --- a/data/protos/task_pb2_grpc.py +++ b/data/protos/task_pb2_grpc.py @@ -1,36 +1,33 @@ # Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT! """Client and server classes corresponding to protobuf-defined services.""" -import warnings - import grpc +import warnings -from . import task_pb2 as task__pb2 +from protos import task_pb2 as protos_dot_task__pb2 -GRPC_GENERATED_VERSION = "1.76.0" +GRPC_GENERATED_VERSION = '1.76.0' GRPC_VERSION = grpc.__version__ _version_not_supported = False try: from grpc._utilities import first_version_is_lower - - _version_not_supported = first_version_is_lower( - GRPC_VERSION, GRPC_GENERATED_VERSION - ) + _version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION) except ImportError: _version_not_supported = True 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" - + 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}." + f'The grpc package installed is at version {GRPC_VERSION},' + + ' 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}.' ) class TaskServiceStub(object): - """TaskService 定义了任务执行与管理的 RPC 接口。""" + """TaskService 定义了任务执行与管理的 RPC 接口。 + """ def __init__(self, channel): """Constructor. @@ -39,112 +36,128 @@ def __init__(self, channel): channel: A grpc.Channel. """ self.CollectBinance = channel.unary_unary( - "/task.v1.TaskService/CollectBinance", - request_serializer=task__pb2.CollectBinanceRequest.SerializeToString, - response_deserializer=task__pb2.TaskResponse.FromString, - _registered_method=True, - ) + '/task.v1.TaskService/CollectBinance', + 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, - _registered_method=True, - ) + '/task.v1.TaskService/CollectUniswap', + 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, - _registered_method=True, - ) + '/task.v1.TaskService/ProcessPrices', + 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, - _registered_method=True, - ) + '/task.v1.TaskService/Analyse', + request_serializer=protos_dot_task__pb2.AnalyseRequest.SerializeToString, + response_deserializer=protos_dot_task__pb2.TaskResponse.FromString, + _registered_method=True) class TaskServiceServicer(object): - """TaskService 定义了任务执行与管理的 RPC 接口。""" + """TaskService 定义了任务执行与管理的 RPC 接口。 + """ def CollectBinance(self, request, context): - """收集币安数据""" + """收集币安数据 + """ + context.set_code(grpc.StatusCode.UNIMPLEMENTED) + 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!") + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') def CollectUniswap(self, request, context): - """收集Uniswap数据""" + """收集Uniswap数据 + """ context.set_code(grpc.StatusCode.UNIMPLEMENTED) - context.set_details("Method not implemented!") - raise NotImplementedError("Method not implemented!") + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') def ProcessPrices(self, request, context): - """处理价格数据""" + """处理价格数据 + """ context.set_code(grpc.StatusCode.UNIMPLEMENTED) - context.set_details("Method not implemented!") - raise NotImplementedError("Method not implemented!") + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') def Analyse(self, request, context): - """分析数据""" + """分析数据 + """ context.set_code(grpc.StatusCode.UNIMPLEMENTED) - context.set_details("Method not implemented!") - raise NotImplementedError("Method not implemented!") + context.set_details('Method not implemented!') + raise NotImplementedError('Method not implemented!') 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, - ), - "CollectUniswap": grpc.unary_unary_rpc_method_handler( - servicer.CollectUniswap, - request_deserializer=task__pb2.CollectUniswapRequest.FromString, - response_serializer=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, - ), - "Analyse": grpc.unary_unary_rpc_method_handler( - servicer.Analyse, - request_deserializer=task__pb2.AnalyseRequest.FromString, - response_serializer=task__pb2.TaskResponse.SerializeToString, - ), + 'CollectBinance': grpc.unary_unary_rpc_method_handler( + servicer.CollectBinance, + 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=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=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=protos_dot_task__pb2.AnalyseRequest.FromString, + response_serializer=protos_dot_task__pb2.TaskResponse.SerializeToString, + ), } generic_handler = grpc.method_handlers_generic_handler( - "task.v1.TaskService", rpc_method_handlers - ) + 'task.v1.TaskService', rpc_method_handlers) server.add_generic_rpc_handlers((generic_handler,)) - server.add_registered_method_handlers("task.v1.TaskService", rpc_method_handlers) + server.add_registered_method_handlers('task.v1.TaskService', rpc_method_handlers) -# This class is part of an EXPERIMENTAL API. + # This class is part of an EXPERIMENTAL API. class TaskService(object): - """TaskService 定义了任务执行与管理的 RPC 接口。""" + """TaskService 定义了任务执行与管理的 RPC 接口。 + """ @staticmethod - def CollectBinance( - request, - target, - options=(), - channel_credentials=None, - call_credentials=None, - insecure=False, - compression=None, - wait_for_ready=None, - timeout=None, - metadata=None, - ): + def CollectBinance(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/CollectBinance", - task__pb2.CollectBinanceRequest.SerializeToString, - task__pb2.TaskResponse.FromString, + '/task.v1.TaskService/CollectBinance', + protos_dot_task__pb2.CollectBinanceRequest.SerializeToString, + protos_dot_task__pb2.TaskResponse.FromString, options, channel_credentials, insecure, @@ -153,28 +166,25 @@ def CollectBinance( wait_for_ready, timeout, metadata, - _registered_method=True, - ) + _registered_method=True) @staticmethod - def CollectUniswap( - request, - target, - options=(), - channel_credentials=None, - call_credentials=None, - insecure=False, - compression=None, - wait_for_ready=None, - timeout=None, - metadata=None, - ): + 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/CollectUniswap", - task__pb2.CollectUniswapRequest.SerializeToString, - task__pb2.TaskResponse.FromString, + '/task.v1.TaskService/CollectBinanceByDate', + protos_dot_task__pb2.CollectBinaceByDateRequest.SerializeToString, + protos_dot_task__pb2.TaskResponse.FromString, options, channel_credentials, insecure, @@ -183,28 +193,25 @@ def CollectUniswap( wait_for_ready, timeout, metadata, - _registered_method=True, - ) + _registered_method=True) @staticmethod - def ProcessPrices( - request, - target, - options=(), - channel_credentials=None, - call_credentials=None, - insecure=False, - compression=None, - wait_for_ready=None, - timeout=None, - metadata=None, - ): + def CollectUniswap(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/ProcessPrices", - task__pb2.ProcessPricesRequest.SerializeToString, - task__pb2.TaskResponse.FromString, + '/task.v1.TaskService/CollectUniswap', + protos_dot_task__pb2.CollectUniswapRequest.SerializeToString, + protos_dot_task__pb2.TaskResponse.FromString, options, channel_credentials, insecure, @@ -213,28 +220,52 @@ def ProcessPrices( wait_for_ready, timeout, metadata, - _registered_method=True, - ) + _registered_method=True) @staticmethod - def Analyse( - request, - target, - options=(), - channel_credentials=None, - call_credentials=None, - insecure=False, - compression=None, - wait_for_ready=None, - timeout=None, - metadata=None, - ): + def ProcessPrices(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/ProcessPrices', + protos_dot_task__pb2.ProcessPricesRequest.SerializeToString, + protos_dot_task__pb2.TaskResponse.FromString, + options, + channel_credentials, + insecure, + call_credentials, + compression, + wait_for_ready, + timeout, + metadata, + _registered_method=True) + + @staticmethod + def Analyse(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/Analyse", - task__pb2.AnalyseRequest.SerializeToString, - task__pb2.TaskResponse.FromString, + '/task.v1.TaskService/Analyse', + protos_dot_task__pb2.AnalyseRequest.SerializeToString, + protos_dot_task__pb2.TaskResponse.FromString, options, channel_credentials, insecure, @@ -243,5 +274,4 @@ def Analyse( wait_for_ready, timeout, metadata, - _registered_method=True, - ) + _registered_method=True) diff --git a/data/server.py b/data/server.py index 1d629f2..038ff8b 100644 --- a/data/server.py +++ b/data/server.py @@ -152,6 +152,51 @@ 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/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; // 池地址 From 739f4081489fe5d030853fdcf9655b4f508e752d Mon Sep 17 00:00:00 2001 From: cagoya <2561694274@qq.com> Date: Wed, 24 Dec 2025 23:58:40 +0800 Subject: [PATCH 2/4] =?UTF-8?q?fix=EF=BC=9A=E4=BF=AE=E6=AD=A3data=E4=B8=AD?= =?UTF-8?q?task=20status=E7=9A=84=E5=90=8D=E7=A7=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- data/block_chain/collect_binance.py | 18 +++++++++--------- data/block_chain/collect_uniswap.py | 4 ++-- data/block_chain/task.py | 2 +- 3 files changed, 12 insertions(+), 12 deletions(-) diff --git a/data/block_chain/collect_binance.py b/data/block_chain/collect_binance.py index 9cd057d..0a7badb 100644 --- a/data/block_chain/collect_binance.py +++ b/data/block_chain/collect_binance.py @@ -60,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 @@ -122,7 +122,7 @@ 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 @@ -173,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 @@ -211,11 +211,11 @@ def collect_binance( 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 @@ -362,13 +362,13 @@ def collect_binance_by_date( logger.info( f"成功导入 {total_rows_imported} 行，耗时 {total_time:.2f}s" ) - update_task_status(task_id, "TASK_STATUS_SUCCESS") + 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, "TASK_STATUS_FAILED") + update_task_status(task_id, "FAILED") raise 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 From d952877ffe2b3a8dcad29a1c62c09d5680d9e725 Mon Sep 17 00:00:00 2001 From: cagoya <2561694274@qq.com> Date: Thu, 25 Dec 2025 00:00:41 +0800 Subject: [PATCH 3/4] =?UTF-8?q?style=EF=BC=9A=E6=A0=BC=E5=BC=8F=E5=8C=96?= =?UTF-8?q?=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .github/workflows/cicd.yaml | 18 +- data/block_chain/collect_binance.py | 64 +++--- data/protos/task_pb2.py | 60 +++--- data/protos/task_pb2_grpc.py | 301 +++++++++++++++------------- data/server.py | 4 +- 5 files changed, 232 insertions(+), 215 deletions(-) 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/data/block_chain/collect_binance.py b/data/block_chain/collect_binance.py index 0a7badb..ce66cfe 100644 --- a/data/block_chain/collect_binance.py +++ b/data/block_chain/collect_binance.py @@ -219,7 +219,9 @@ def collect_binance( raise -def download_binance_file(task_id: str, date_str: str, symbol: str = "ETHUSDT") -> Optional[str]: +def download_binance_file( + task_id: str, date_str: str, symbol: str = "ETHUSDT" +) -> Optional[str]: """ 描述：从币安数据源下载指定日期的交易数据文件 参数：task_id: 任务ID, date_str: 日期字符串 (YYYY-MM-DD), symbol: 交易对符号 @@ -227,16 +229,16 @@ def download_binance_file(task_id: str, date_str: str, symbol: str = "ETHUSDT") """ 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): @@ -247,7 +249,7 @@ def download_binance_file(task_id: str, date_str: str, symbol: str = "ETHUSDT") os.remove(zip_path) os.rmdir(temp_dir) return None - + # 解压文件 csv_path = None with zipfile.ZipFile(zip_path, "r") as zip_ref: @@ -258,17 +260,17 @@ def download_binance_file(task_id: str, date_str: str, symbol: str = "ETHUSDT") 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 @@ -281,7 +283,11 @@ def download_binance_file(task_id: str, date_str: str, symbol: str = "ETHUSDT") def collect_binance_by_date( - task_id: str, start_ts: int, end_ts: int, symbol: str = "ETHUSDT", chunk_size: int = 1000000 + task_id: str, + start_ts: int, + end_ts: int, + symbol: str = "ETHUSDT", + chunk_size: int = 1000000, ) -> int: """ 描述：按日期范围收集币安数据 @@ -295,41 +301,39 @@ def collect_binance_by_date( """ 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()}" - ) - + + 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( @@ -340,7 +344,7 @@ def collect_binance_by_date( except Exception as e: logger.error(f"导入日期 {date_str} 的数据失败: {e}") # 继续处理下一个日期，不中断整个任务 - + # 清理临时文件 try: if os.path.exists(csv_path): @@ -350,21 +354,19 @@ def collect_binance_by_date( 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" - ) + + 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) diff --git a/data/protos/task_pb2.py b/data/protos/task_pb2.py index 6a94914..3eb0009 100644 --- a/data/protos/task_pb2.py +++ b/data/protos/task_pb2.py @@ -9,47 +9,43 @@ from google.protobuf import runtime_version as _runtime_version from google.protobuf import symbol_database as _symbol_database from google.protobuf.internal import builder as _builder + _runtime_version.ValidateProtobufRuntimeVersion( - _runtime_version.Domain.PUBLIC, - 6, - 31, - 1, - '', - 'protos/task.proto' + _runtime_version.Domain.PUBLIC, 6, 31, 1, "", "protos/task.proto" ) # @@protoc_insertion_point(imports) _sym_db = _symbol_database.Default() - - -DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile(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') +DESCRIPTOR = _descriptor_pool.Default().AddSerializedFile( + 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, 'protos.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=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 + _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 = 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 29268ce..b32ca65 100644 --- a/data/protos/task_pb2_grpc.py +++ b/data/protos/task_pb2_grpc.py @@ -1,33 +1,36 @@ # Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT! """Client and server classes corresponding to protobuf-defined services.""" -import grpc import warnings +import grpc + from protos import task_pb2 as protos_dot_task__pb2 -GRPC_GENERATED_VERSION = '1.76.0' +GRPC_GENERATED_VERSION = "1.76.0" GRPC_VERSION = grpc.__version__ _version_not_supported = False try: from grpc._utilities import first_version_is_lower - _version_not_supported = first_version_is_lower(GRPC_VERSION, GRPC_GENERATED_VERSION) + + _version_not_supported = first_version_is_lower( + GRPC_VERSION, GRPC_GENERATED_VERSION + ) except ImportError: _version_not_supported = True if _version_not_supported: raise RuntimeError( - f'The grpc package installed is at version {GRPC_VERSION},' - + ' 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}.' + f"The grpc package installed is at version {GRPC_VERSION}," + + " 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}." ) class TaskServiceStub(object): - """TaskService 定义了任务执行与管理的 RPC 接口。 - """ + """TaskService 定义了任务执行与管理的 RPC 接口。""" def __init__(self, channel): """Constructor. @@ -36,126 +39,127 @@ def __init__(self, channel): channel: A grpc.Channel. """ self.CollectBinance = channel.unary_unary( - '/task.v1.TaskService/CollectBinance', - request_serializer=protos_dot_task__pb2.CollectBinanceRequest.SerializeToString, - response_deserializer=protos_dot_task__pb2.TaskResponse.FromString, - _registered_method=True) + "/task.v1.TaskService/CollectBinance", + 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) + "/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=protos_dot_task__pb2.CollectUniswapRequest.SerializeToString, - response_deserializer=protos_dot_task__pb2.TaskResponse.FromString, - _registered_method=True) + "/task.v1.TaskService/CollectUniswap", + 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=protos_dot_task__pb2.ProcessPricesRequest.SerializeToString, - response_deserializer=protos_dot_task__pb2.TaskResponse.FromString, - _registered_method=True) + "/task.v1.TaskService/ProcessPrices", + 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=protos_dot_task__pb2.AnalyseRequest.SerializeToString, - response_deserializer=protos_dot_task__pb2.TaskResponse.FromString, - _registered_method=True) + "/task.v1.TaskService/Analyse", + request_serializer=protos_dot_task__pb2.AnalyseRequest.SerializeToString, + response_deserializer=protos_dot_task__pb2.TaskResponse.FromString, + _registered_method=True, + ) class TaskServiceServicer(object): - """TaskService 定义了任务执行与管理的 RPC 接口。 - """ + """TaskService 定义了任务执行与管理的 RPC 接口。""" def CollectBinance(self, request, context): - """收集币安数据 - """ + """收集币安数据""" context.set_code(grpc.StatusCode.UNIMPLEMENTED) - context.set_details('Method not implemented!') - raise NotImplementedError('Method not implemented!') + 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!') + context.set_details("Method not implemented!") + raise NotImplementedError("Method not implemented!") def CollectUniswap(self, request, context): - """收集Uniswap数据 - """ + """收集Uniswap数据""" context.set_code(grpc.StatusCode.UNIMPLEMENTED) - context.set_details('Method not implemented!') - raise NotImplementedError('Method not implemented!') + context.set_details("Method not implemented!") + raise NotImplementedError("Method not implemented!") def ProcessPrices(self, request, context): - """处理价格数据 - """ + """处理价格数据""" context.set_code(grpc.StatusCode.UNIMPLEMENTED) - context.set_details('Method not implemented!') - raise NotImplementedError('Method not implemented!') + context.set_details("Method not implemented!") + raise NotImplementedError("Method not implemented!") def Analyse(self, request, context): - """分析数据 - """ + """分析数据""" context.set_code(grpc.StatusCode.UNIMPLEMENTED) - context.set_details('Method not implemented!') - raise NotImplementedError('Method not implemented!') + context.set_details("Method not implemented!") + raise NotImplementedError("Method not implemented!") def add_TaskServiceServicer_to_server(servicer, server): rpc_method_handlers = { - 'CollectBinance': grpc.unary_unary_rpc_method_handler( - servicer.CollectBinance, - 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=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=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=protos_dot_task__pb2.AnalyseRequest.FromString, - response_serializer=protos_dot_task__pb2.TaskResponse.SerializeToString, - ), + "CollectBinance": grpc.unary_unary_rpc_method_handler( + servicer.CollectBinance, + 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=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=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=protos_dot_task__pb2.AnalyseRequest.FromString, + response_serializer=protos_dot_task__pb2.TaskResponse.SerializeToString, + ), } generic_handler = grpc.method_handlers_generic_handler( - 'task.v1.TaskService', rpc_method_handlers) + "task.v1.TaskService", rpc_method_handlers + ) server.add_generic_rpc_handlers((generic_handler,)) - server.add_registered_method_handlers('task.v1.TaskService', rpc_method_handlers) + server.add_registered_method_handlers("task.v1.TaskService", rpc_method_handlers) - # This class is part of an EXPERIMENTAL API. +# This class is part of an EXPERIMENTAL API. class TaskService(object): - """TaskService 定义了任务执行与管理的 RPC 接口。 - """ + """TaskService 定义了任务执行与管理的 RPC 接口。""" @staticmethod - def CollectBinance(request, - target, - options=(), - channel_credentials=None, - call_credentials=None, - insecure=False, - compression=None, - wait_for_ready=None, - timeout=None, - metadata=None): + def CollectBinance( + 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/CollectBinance', + "/task.v1.TaskService/CollectBinance", protos_dot_task__pb2.CollectBinanceRequest.SerializeToString, protos_dot_task__pb2.TaskResponse.FromString, options, @@ -166,23 +170,26 @@ def CollectBinance(request, wait_for_ready, timeout, metadata, - _registered_method=True) + _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): + 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', + "/task.v1.TaskService/CollectBinanceByDate", protos_dot_task__pb2.CollectBinaceByDateRequest.SerializeToString, protos_dot_task__pb2.TaskResponse.FromString, options, @@ -193,23 +200,26 @@ def CollectBinanceByDate(request, wait_for_ready, timeout, metadata, - _registered_method=True) + _registered_method=True, + ) @staticmethod - def CollectUniswap(request, - target, - options=(), - channel_credentials=None, - call_credentials=None, - insecure=False, - compression=None, - wait_for_ready=None, - timeout=None, - metadata=None): + def CollectUniswap( + 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/CollectUniswap', + "/task.v1.TaskService/CollectUniswap", protos_dot_task__pb2.CollectUniswapRequest.SerializeToString, protos_dot_task__pb2.TaskResponse.FromString, options, @@ -220,23 +230,26 @@ def CollectUniswap(request, wait_for_ready, timeout, metadata, - _registered_method=True) + _registered_method=True, + ) @staticmethod - def ProcessPrices(request, - target, - options=(), - channel_credentials=None, - call_credentials=None, - insecure=False, - compression=None, - wait_for_ready=None, - timeout=None, - metadata=None): + def ProcessPrices( + 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/ProcessPrices', + "/task.v1.TaskService/ProcessPrices", protos_dot_task__pb2.ProcessPricesRequest.SerializeToString, protos_dot_task__pb2.TaskResponse.FromString, options, @@ -247,23 +260,26 @@ def ProcessPrices(request, wait_for_ready, timeout, metadata, - _registered_method=True) + _registered_method=True, + ) @staticmethod - def Analyse(request, - target, - options=(), - channel_credentials=None, - call_credentials=None, - insecure=False, - compression=None, - wait_for_ready=None, - timeout=None, - metadata=None): + def Analyse( + 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/Analyse', + "/task.v1.TaskService/Analyse", protos_dot_task__pb2.AnalyseRequest.SerializeToString, protos_dot_task__pb2.TaskResponse.FromString, options, @@ -274,4 +290,5 @@ def Analyse(request, wait_for_ready, timeout, metadata, - _registered_method=True) + _registered_method=True, + ) diff --git a/data/server.py b/data/server.py index 038ff8b..86a2660 100644 --- a/data/server.py +++ b/data/server.py @@ -180,7 +180,9 @@ def run_task(): logger.info(f"任务 {task_id} 执行成功: 按日期收集币安数据") log_task_event(task_id, "INFO", "按日期收集 Binance 数据完成") mark_task_finished( - task_id, TaskStatus.TASK_STATUS_SUCCESS, "Binance 数据按日期收集完成" + task_id, + TaskStatus.TASK_STATUS_SUCCESS, + "Binance 数据按日期收集完成", ) except Exception as e: logger.error(f"任务 {task_id} 执行失败: {e}") From 090a9ec3922f28e6b18583beb5e9ba1930fff9f3 Mon Sep 17 00:00:00 2001 From: cagoya <2561694274@qq.com> Date: Thu, 25 Dec 2025 00:05:40 +0800 Subject: [PATCH 4/4] =?UTF-8?q?fix=EF=BC=9A=E4=BF=AE=E5=A4=8D=E5=8D=95?= =?UTF-8?q?=E5=85=83=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- data/test/test_collect_binance.py | 18 +++++++++--------- data/test/test_task.py | 2 +- 2 files changed, 10 insertions(+), 10 deletions(-) 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")