Alpha Common

Alpha 策略研究基础设施库，提供量化研究中常用的四个核心模块：

• xcals — A 股交易日历（交易日查询、日期偏移、报告期计算）
• blazestore — 本地 Parquet 存储引擎（Hive 分区、SQL 查询、MySQL/ClickHouse 客户端）
• ygo — 并发任务框架（延迟执行、线程池、进度条）
• clickhouse_df — ClickHouse 数据库驱动（Polars/Pandas 读写、命令行批量下载）

安装

pip install alpha-common

开发模式（使用 uv）：

git clone https://github.com/Aequiludium/alpha-common.git
cd alpha-common
uv sync

使用

xcals — 交易日历

import xcals

# 交易日查询
days = xcals.get_tradingdays("2024-01-01", "2024-12-31")
# -> ['2024-01-02', '2024-01-03', ..., '2024-12-31']

xcals.is_tradeday("2024-12-31")  # True

# 日期偏移（非交易日自动跳到最近交易日）
prev = xcals.shift_tradeday("2024-12-31", -5)
# -> '2024-12-24'

# 最近交易日
xcals.get_last_tradingday("2024-01-01")
# -> '2023-12-29'

# 报告期计算（获取前 2 个季报截止日）
xcals.get_previous_report_dates("2024-10-15", n=2)
# -> ['2024-06-30', '2024-09-30']

# n=1 时返回单个报告期
xcals.get_previous_report_dates("2024-10-15", n=1)
# -> '2024-09-30'

# 更新交易日数据（从远程下载最新日历）
xcals.update()

blazestore — 本地 Parquet 存储引擎

支持三种写入模式和 SQL 查询，配置路径默认为 ~/.blaze/config.toml。

模块级 API（推荐）

from blazestore import put, read, sql, list_tables

# 写入：自动识别模式
put(df, "trades")                          # 写入 trades/data.parquet
put(df, "trades/2024.parquet")             # 直接写入文件
put(df, "trades", partitions=["date"])     # Hive 分区写入

# 读取（返回 LazyFrame，自动识别 Hive 分区）
lf = read("trades")
df = lf.filter(pl.col("symbol") == "AAPL").collect()

# 对本地 Parquet 文件执行 SQL 查询
result = sql("SELECT date, count(*) FROM trades GROUP BY date")

类 API

from blazestore import ParquetStore

store = ParquetStore("/data/store")

# 写入
store.put(df, "trades")
store.put(df, "trades", partitions=["date"])

# 读取
lf = store.read("trades")

# 表管理
store.list_tables()              # -> ['trades', 'orders']
store.get_table_info("trades")   # -> {'name': 'trades', 'rows': 1000, ...}
store.optimize_table("trades")   # 合并小文件
store.check_table("trades")      # -> True
store.delete_table("old_table")

数据库客户端

from blazestore import read_ck, read_mysql, write_mysql, download_ck

# 从 ClickHouse 读取
df = read_ck("SELECT * FROM trades WHERE date = '2024-01-01'")

# 从 MySQL 读取
df = read_mysql("SELECT * FROM users WHERE id = 1")

# 写入 MySQL
write_mysql(df, "users")

# ClickHouse 批量下载到文件（使用 clickhouse-client）
download_ck("SELECT * FROM big_table", "output.parquet")

配置示例（~/.blaze/config.toml）：

[paths]
store = "/home/user/BlazeStore"

[databases.ck]
urls = ["192.168.1.100:9000"]
user = "default"
password = ""

[databases.mysql]
url = "127.0.0.1:3306"
user = "root"
password = ""

ygo — 并发任务框架

基于 joblib 的并行调度，支持任务分组、进度条。

from ygo import Pool

pool = Pool(n_jobs=4, show_progress=True)

def download(date: str) -> dict:
    return {"date": date, "data": fetch_data(date)}

# submit 返回任务收集函数；调用它会将任务加入池中
queue_download = pool.submit(download, job_name="download")
queue_download(date="2024-01-01")
queue_download(date="2024-01-02")

# 并行执行所有任务
results = pool.do()  # -> [{"date": "2024-01-01", ...}, {"date": "2024-01-02", ...}]

# 方式二：延迟函数（适用于批量生成）
from ygo import delay

jobs = [delay(fetch_data).bind(day=d) for d in trading_days]
pool.submit_batch(jobs, job_name="batch_download")
pool.do()

Pool 支持上下文管理器：

with Pool(n_jobs=8) as pool:
    for day in trading_days:
        pool.submit(download)(date=day)
    results = pool.do()

show_progress=True 保持为默认值，并在业务终端显示一条紧凑的聚合进度。 本机监控默认开启，与内联进度相互独立；需要安静执行时可以使用 show_progress=False，需要完全关闭监控状态发布时使用 monitor=False 或设置 YGO_MONITOR=0。

在另一个终端中使用增量刷新的任务监控器：

ygo top                       # 交互式监控所有本机 ygo 任务
ygo ps                        # 输出一次当前任务列表
ygo show <pool-id>            # 查看一个 Pool
ygo errors <pool-id>          # 查看当前错误摘要
ygo run -- python script.py   # 运行可被 ygo top 发现的程序

ygo top 使用共享内存读取聚合状态，只更新发生变化的表格单元格。它显示所有 实时任务组和最近完成的 100 个任务组；完成记录保存在平台用户状态目录中，不会 随生产进程退出而清理。STARTED 使用本机时间，默认让最新任务排在顶端。点击 任意列头可按该字段排序，再次点击可切换升序和降序。业务日志仍保留在原程序 终端，不会破坏监控界面。

clickhouse_df — ClickHouse 数据库驱动

import clickhouse_df

# 连接（随机负载均衡）
conn = clickhouse_df.connect(
    urls=["192.168.1.100:9000", "192.168.1.101:9000"],
    user="default",
    password="",
)

# 查询为 Polars DataFrame
df = clickhouse_df.to_polars("SELECT * FROM trades LIMIT 10")
# shape: (10, 5)

# 查询为 Pandas DataFrame
pdf = clickhouse_df.to_pandas("SELECT * FROM trades LIMIT 10")

# 关闭当前线程所有连接
clickhouse_df.close_all()

# 命令行批量下载（适合大结果集，直接写入 Parquet）
clickhouse_df.raw_download("SELECT * FROM big_table", "output.parquet", settings)

开发

uv sync                    # 安装依赖
uv run pytest tests/       # 运行测试
uv run ruff check .        # 代码检查
uv run ruff format .       # 格式化代码

许可证

MIT