跳转至

cryptoservice.interfaces.market_interfaces

cryptoservice.interfaces.market_interfaces

Classes

IMarketDataService

Bases: Protocol

市场数据服务接口

Functions
get_market_summary(interval: Freq = Freq.d1) -> dict[str, Any]

获取市场概况.

PARAMETER DESCRIPTION
symbols

交易对列表
interval

数据间隔

TYPE: Freq DEFAULT: d1

RETURNS DESCRIPTION
dict[str, Any]

Dict[str, Any]: 市场概况数据
Source code in src/cryptoservice/interfaces/market_interfaces.py 
41
42
43
44
45
46
47
48
49
50
51
def get_market_summary(self, interval: Freq = Freq.d1) -> dict[str, Any]:
    """获取市场概况.

    Args:
        symbols: 交易对列表
        interval: 数据间隔

    Returns:
        Dict[str, Any]: 市场概况数据
    """
    ...
get_symbol_ticker(symbol: str | None = None) -> SymbolTicker | list[SymbolTicker] abstractmethod

获取单个或多个交易币的行情数据.

PARAMETER DESCRIPTION
symbol

交易对名称，如果为 None 则返回所有交易对数据

TYPE: str | None DEFAULT: None

RETURNS DESCRIPTION
SymbolTicker | list[SymbolTicker]
  • 当 symbol 指定时：返回单个 SymbolTicker
SymbolTicker | list[SymbolTicker]
  • 当 symbol 为 None 时：返回 SymbolTicker 列表
RAISES DESCRIPTION
InvalidSymbolError

当指定的交易对不存在时
Source code in src/cryptoservice/interfaces/market_interfaces.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
@abstractmethod
def get_symbol_ticker(self, symbol: str | None = None) -> SymbolTicker | list[SymbolTicker]:
    """获取单个或多个交易币的行情数据.

    Args:
        symbol: 交易对名称，如果为 None 则返回所有交易对数据

    Returns:
        - 当 symbol 指定时：返回单个 SymbolTicker
        - 当 symbol 为 None 时：返回 SymbolTicker 列表

    Raises:
        InvalidSymbolError: 当指定的交易对不存在时
    """
    pass
get_historical_klines(symbol: str, start_time: str | datetime, end_time: str | datetime | None = None, interval: Freq = Freq.d1) -> list[KlineMarketTicker] abstractmethod

获取历史行情数据.

PARAMETER DESCRIPTION
symbol

交易对名称

TYPE: str

start_time

开始时间

TYPE: str | datetime

end_time

结束时间，默认为当前时间

TYPE: str | datetime | None DEFAULT: None

interval

数据间隔，如 1m, 5m, 1h, 1d

TYPE: Freq DEFAULT: d1

RETURNS DESCRIPTION
list[KlineMarketTicker]

List[KlineMarketTicker]: 历史行情数据列表
Source code in src/cryptoservice/interfaces/market_interfaces.py 
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
@abstractmethod
def get_historical_klines(
    self,
    symbol: str,
    start_time: str | datetime,
    end_time: str | datetime | None = None,
    interval: Freq = Freq.d1,
) -> list[KlineMarketTicker]:
    """获取历史行情数据.

    Args:
        symbol: 交易对名称
        start_time: 开始时间
        end_time: 结束时间，默认为当前时间
        interval: 数据间隔，如 1m, 5m, 1h, 1d

    Returns:
        List[KlineMarketTicker]: 历史行情数据列表
    """
    pass
get_perpetual_data(symbols: list[str], start_time: str, db_path: Path | str, end_time: str | None = None, interval: Freq = Freq.h1, max_workers: int = 5, max_retries: int = 3, progress: Progress | None = None, request_delay: float = 0.5, retry_config: Optional[RetryConfig] = None, enable_integrity_check: bool = True) -> IntegrityReport abstractmethod

获取永续合约历史数据, 并存储到指定数据库.

PARAMETER DESCRIPTION
symbols

交易对列表

TYPE: list[str]

start_time

开始时间 (YYYYMMDD)

TYPE: str

end_time

结束时间 (YYYYMMDD)

TYPE: str | None DEFAULT: None

interval

数据频率 (1m, 1h, 4h, 1d等)

TYPE: Freq DEFAULT: h1

db_path

数据库文件路径

TYPE: Path | str

max_workers

并发线程数

TYPE: int DEFAULT: 5

max_retries

最大重试次数

TYPE: int DEFAULT: 3

progress

进度条

TYPE: Progress | None DEFAULT: None

request_delay

每次请求间隔（秒）

TYPE: float DEFAULT: 0.5

Source code in src/cryptoservice/interfaces/market_interfaces.py 
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
@abstractmethod
def get_perpetual_data(
    self,
    symbols: list[str],
    start_time: str,
    db_path: Path | str,
    end_time: str | None = None,
    interval: Freq = Freq.h1,
    max_workers: int = 5,
    max_retries: int = 3,
    progress: Progress | None = None,
    request_delay: float = 0.5,
    retry_config: Optional[RetryConfig] = None,
    enable_integrity_check: bool = True,
) -> IntegrityReport:
    """获取永续合约历史数据, 并存储到指定数据库.

    Args:
        symbols: 交易对列表
        start_time: 开始时间 (YYYYMMDD)
        end_time: 结束时间 (YYYYMMDD)
        interval: 数据频率 (1m, 1h, 4h, 1d等)
        db_path: 数据库文件路径
        max_workers: 并发线程数
        max_retries: 最大重试次数
        progress: 进度条
        request_delay: 每次请求间隔（秒）

    """
    pass
download_universe_data(universe_file: Path | str, db_path: Path | str, data_path: Path | str | None = None, interval: Freq = Freq.h1, max_workers: int = 4, max_retries: int = 3, include_buffer_days: int = 7, retry_config: Optional[RetryConfig] = None, request_delay: float = 0.5) -> None abstractmethod

根据universe定义文件下载相应的历史数据到数据库.

PARAMETER DESCRIPTION
universe_file

universe定义文件路径

TYPE: Path | str

db_path

数据库文件路径

TYPE: Path | str

data_path

数据文件存储路径 (可选)

TYPE: Path | str | None DEFAULT: None

interval

数据频率 (1m, 1h, 4h, 1d等)

TYPE: Freq DEFAULT: h1

max_workers

并发线程数

TYPE: int DEFAULT: 4

max_retries

最大重试次数

TYPE: int DEFAULT: 3

include_buffer_days

在数据期间前后增加的缓冲天数

TYPE: int DEFAULT: 7

request_delay

每次请求间隔（秒）

TYPE: float DEFAULT: 0.5

Source code in src/cryptoservice/interfaces/market_interfaces.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
@abstractmethod
def download_universe_data(
    self,
    universe_file: Path | str,
    db_path: Path | str,
    data_path: Path | str | None = None,
    interval: Freq = Freq.h1,
    max_workers: int = 4,
    max_retries: int = 3,
    include_buffer_days: int = 7,
    retry_config: Optional[RetryConfig] = None,
    request_delay: float = 0.5,
) -> None:
    """根据universe定义文件下载相应的历史数据到数据库.

    Args:
        universe_file: universe定义文件路径
        db_path: 数据库文件路径
        data_path: 数据文件存储路径 (可选)
        interval: 数据频率 (1m, 1h, 4h, 1d等)
        max_workers: 并发线程数
        max_retries: 最大重试次数
        include_buffer_days: 在数据期间前后增加的缓冲天数
        request_delay: 每次请求间隔（秒）

    """
    pass
define_universe(start_date: str, end_date: str, t1_months: int, t2_months: int, t3_months: int, output_path: Path | str, top_k: int | None = None, top_ratio: float | None = None, description: str | None = None, delay_days: int = 7, api_delay_seconds: float = 1.0, batch_delay_seconds: float = 3.0, batch_size: int = 5, quote_asset: str = 'USDT') -> UniverseDefinition abstractmethod

定义universe并保存到文件.

PARAMETER DESCRIPTION
start_date

开始日期 (YYYY-MM-DD 或 YYYYMMDD)

TYPE: str

end_date

结束日期 (YYYY-MM-DD 或 YYYYMMDD)

TYPE: str

t1_months

T1时间窗口（月），用于计算mean daily amount

TYPE: int

t2_months

T2滚动频率（月），universe重新选择的频率

TYPE: int

t3_months

T3合约最小创建时间（月），用于筛除新合约

TYPE: int

output_path

universe输出文件路径 (必须指定)

TYPE: Path | str

top_k

选取的top合约数量 (与 top_ratio 二选一)

TYPE: int | None DEFAULT: None

top_ratio

选取的top合约比率 (与 top_k 二选一)

TYPE: float | None DEFAULT: None

description

描述信息

TYPE: str | None DEFAULT: None

delay_days

在重新平衡日期前额外往前推的天数，默认7天

TYPE: int DEFAULT: 7

api_delay_seconds

每个API请求之间的延迟秒数，默认1.0秒

TYPE: float DEFAULT: 1.0

batch_delay_seconds

每批次请求之间的延迟秒数，默认3.0秒

TYPE: float DEFAULT: 3.0

batch_size

每批次的请求数量，默认5个

TYPE: int DEFAULT: 5

quote_asset

基准资产，默认为USDT，只筛选以该资产结尾的交易对

TYPE: str DEFAULT: 'USDT'

RETURNS DESCRIPTION
UniverseDefinition

定义的universe

TYPE: UniverseDefinition

Source code in src/cryptoservice/interfaces/market_interfaces.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
@abstractmethod
def define_universe(
    self,
    start_date: str,
    end_date: str,
    t1_months: int,
    t2_months: int,
    t3_months: int,
    output_path: Path | str,
    top_k: int | None = None,
    top_ratio: float | None = None,
    description: str | None = None,
    delay_days: int = 7,
    api_delay_seconds: float = 1.0,
    batch_delay_seconds: float = 3.0,
    batch_size: int = 5,
    quote_asset: str = "USDT",
) -> UniverseDefinition:
    """定义universe并保存到文件.

    Args:
        start_date: 开始日期 (YYYY-MM-DD 或 YYYYMMDD)
        end_date: 结束日期 (YYYY-MM-DD 或 YYYYMMDD)
        t1_months: T1时间窗口（月），用于计算mean daily amount
        t2_months: T2滚动频率（月），universe重新选择的频率
        t3_months: T3合约最小创建时间（月），用于筛除新合约
        output_path: universe输出文件路径 (必须指定)
        top_k: 选取的top合约数量 (与 top_ratio 二选一)
        top_ratio: 选取的top合约比率 (与 top_k 二选一)
        description: 描述信息
        delay_days: 在重新平衡日期前额外往前推的天数，默认7天
        api_delay_seconds: 每个API请求之间的延迟秒数，默认1.0秒
        batch_delay_seconds: 每批次请求之间的延迟秒数，默认3.0秒
        batch_size: 每批次的请求数量，默认5个
        quote_asset: 基准资产，默认为USDT，只筛选以该资产结尾的交易对

    Returns:
        UniverseDefinition: 定义的universe
    """
    pass