Skip to content

数据

数据层目前分成两层：

  • 接口层：zelqor.data.interfaces
  • 具体实现：zelqor.data.clickhouse

也就是说：

  • interfaces.py 负责定义稳定的 provider 协议
  • clickhouse.py 负责真正读取 ClickHouse 数据

当前已经固定了两类返回结果：

  • ProviderFrameBars
  • dict[str, pandas.DataFrame]
  • ProviderNumpyBars
  • dict[str, BarSeries]

另外，全市场批量查询接口会直接返回整张表格式结果：

  • get_market_bars(...)
  • 默认返回 polars.DataFrame
  • 可切换为 pandas.DataFrame
  • get_market_daily_bars(...)
  • 默认返回 polars.DataFrame
  • 可切换为 pandas.DataFrame

并提供了这些辅助函数：

  • is_frame_bars
  • is_numpy_bars
  • require_frame_bars
  • require_numpy_bars

当前 ClickHouseProvider 还会在关键节点输出日志，包括：

  • provider 初始化
  • 交易日查询
  • universe 查询
  • get_bars(...)
  • get_daily_bars(...)

日志级别由：

  • ZELQOR_LOG_LEVEL

控制。

其中 get_trading_dates(...) 现在会先检查 stock_day 的最大可用日期，再用这个日期截断你传入的 end_date。 因此返回的交易日列表不会超出当前实际可用的日线数据范围。

如果你只是想取股票日线数据，当前推荐直接使用 ClickHouseProvider：

from zelqor.models import ClickHouseConfig
from zelqor.data.clickhouse import ClickHouseProvider

config = ClickHouseConfig.from_file("config.toml")
provider = ClickHouseProvider(config)

bars = provider.get_daily_bars(
    codes=["000001.SZ"],
    start_date="2025-03-01",
    end_date="2025-03-25",
    as_numpy=False,
)

frame = bars["000001"]

如果你要取“截至某天最近 N 根日线”：

bars = provider.get_bars(
    codes=["sh600000"],
    end_date="2025-03-25",
    count=20,
    as_numpy=True,
)

series = bars["600000"]

如果你要先把某个交易日的全市场日线整体拉回本地，再做批量筛选：

market_frame = provider.get_market_bars(
    trade_date="2025-03-25",
    as_polars=True,
)

如果你要拉取某个区间的全市场日线：

market_frame = provider.get_market_daily_bars(
    start_date="2025-03-24",
    end_date="2025-03-28",
    as_polars=True,
)

Provider 接口

zelqor.data.interfaces

MarketFrame = MarketPolarsFrame | MarketPandasFrame module-attribute

全市场批量查询可能返回的表格结果。

MarketPandasFrame = pd.DataFrame module-attribute

全市场批量查询返回的 pandas.DataFrame。

MarketPolarsFrame = pl.DataFrame module-attribute

全市场批量查询返回的 polars.DataFrame。

ProviderBars = ProviderFrameBars | ProviderNumpyBars module-attribute

provider 可能返回的 bars 结果联合类型。

ProviderFrameBars = dict[str, pd.DataFrame] module-attribute

按股票代码分组的 pandas.DataFrame 结果。

ProviderNumpyBars = dict[str, BarSeries] module-attribute

按股票代码分组的结构化 NumPy 结果。

MarketDataProvider

Bases: Protocol

统一的行情数据 provider 协议。

get_bars(codes, end_date, count, *, as_numpy=False) 

get_bars(codes: list[str], end_date: str, count: int, *, as_numpy: Literal[False] = False) -> ProviderFrameBars

get_bars(codes: list[str], end_date: str, count: int, *, as_numpy: Literal[True]) -> ProviderNumpyBars

返回截至 end_date 的最近 count 根日线。

get_daily_bars(codes, start_date, end_date, *, as_numpy=True) 

get_daily_bars(codes: list[str], start_date: str, end_date: str, *, as_numpy: Literal[False]) -> ProviderFrameBars

get_daily_bars(codes: list[str], start_date: str, end_date: str, *, as_numpy: Literal[True] = True) -> ProviderNumpyBars

返回指定区间内的全部日线。

get_market_bars(trade_date, *, as_polars=True) 

get_market_bars(trade_date: str, *, as_polars: Literal[True] = True) -> MarketPolarsFrame

get_market_bars(trade_date: str, *, as_polars: Literal[False]) -> MarketPandasFrame

返回指定交易日的全市场日线。

get_market_daily_bars(start_date, end_date, *, as_polars=True) 

get_market_daily_bars(start_date: str, end_date: str, *, as_polars: Literal[True] = True) -> MarketPolarsFrame

get_market_daily_bars(start_date: str, end_date: str, *, as_polars: Literal[False]) -> MarketPandasFrame

返回指定区间内的全市场日线。

get_next_trading_date(trade_date, offset=1)

返回给定交易日之后的第 offset 个交易日。

get_trading_dates(start_date, end_date)

返回指定区间内的交易日列表。

get_universe(trade_date)

返回指定交易日可用的股票 universe。

is_frame_bars(value)

判断 provider 结果是否为 pandas.DataFrame 形式。

is_numpy_bars(value)

判断 provider 结果是否为 BarSeries 形式。

require_frame_bars(value)

将 provider 结果收窄为 pandas.DataFrame 形式。

require_numpy_bars(value)

将 provider 结果收窄为 BarSeries 形式。

ClickHouse Provider

zelqor.data.clickhouse

ClickHouseProvider

基于 ClickHouse 的历史行情读取实现。