使用示例
import os
from datetime import datetime
import polars as pl
os.environ["LOGURU_LEVEL"] = "TRACE"
from datahub import *
def main():
starrocks_setting = StarRocksSetting(
host="xxx",
db_port=9030,
http_port=8040,
username="xxx",
password="xxx",
# sftp=sftp_setting # 设置sftp后会启用大查询缓存
)
setting = Setting(
starrocks=starrocks_setting,
)
datahub = DataHub(setting)
start_time = datetime(2024, 12, 19)
end_time = datetime(2024, 12, 29)
# pl.Config.set_tbl_cols(-1) # -1 表示显示所有列
# pl.Config.set_fmt_str_lengths(100) # 设置字符串显示长度
# pl.Config.set_tbl_width_chars(200) # 设置表格宽度字符数
# 发送监控报警
# mylogger = logger.Logger("test")
# mylogger.monitor.to("alpha运维").info("test")
# mylogger.monitor.to("alpha运维").at("all").error("test")
print("获取交易日")
result = datahub.get_trading_days(start_time, end_time)
print(result)
print("获取指定日期, 指定indicator 可用标的")
result = datahub.get_instrument_list(
trade_date=start_time.date(),
indicators=["premiums_income"],
)
print(result)
print("获取指定日期标的信息")
result = datahub.get_instrument_info(
trade_date=start_time.date(), market="XSHG", instrument_type="spot")
print(result)
print("获取指定日期标的池")
result = datahub.get_universe(trade_date=datetime(
2025, 2, 20), universe="basic_alpha")
print(result)
print("获取某日逐笔成交")
result = datahub.get_md_transaction(
start_date=datetime(2025, 1, 27).date(),
instruments=["508086.XSHG"],
)
print(result)
print("获取某日快照行情")
result = datahub.get_md_snapshot(
start_date=datetime(2025, 1, 27).date(),
instruments=["508086.XSHG"],
)
print(result)
print("获取某日快照行情")
result = datahub.get_trading_days(
start_date=start_time.date(),
end_date=end_time.date(),
)
print(result)
print("获取指标值")
result = datahub.get_indicator_data(
start_time=start_time,
end_time=end_time,
# indicators=["5min_stat_open"],
instruments=["600519.XSHG"],
types=["5min_stat"]
)
print(result)
print("获取BarDataMatrix")
bar_data = datahub.get_indicator_matrix(
trade_time=datetime(2024, 12, 19, 10),
indicators=["5min_stat_open", "5min_stat_low"],
instrument_ids=["600519.XSHG", "000001.XSHE", "000002.XSHG"],
)
print(bar_data)
print("获取BarDataMatrix列表")
bar_data = datahub.get_indicator_matrix_list(
start_date=datetime(2024, 12, 19, 10),
end_date=datetime(2024, 12, 20, 10),
indicators=["5min_stat_open", "5min_stat_low"],
instrument_ids=["600519.XSHG", "000001.XSHE", "000002.XSHG"],
)
print(bar_data)
print("获取因子值")
result = datahub.get_factor_data(
start_time=start_time,
end_time=end_time,
# factors=["5min_stat_open"],
instruments=["600519.XSHG"],
types=["5min_stat"]
)
print(result)
print("获取因子值矩阵")
result = datahub.get_factor_matrix(
trade_time=datetime(2024, 12, 19, 10),
factors=["5min_stat_open", "5min_stat_low"],
instrument_ids=["600519.XSHG", "000001.XSHE", "000002.XSHG"],
)
print(result)
print("获取收益率数据")
result = datahub.get_return_data(
start_time=start_time,
end_time=end_time,
instruments=["600519.XSHG"],
)
print(result)
print("获取收益率矩阵")
result = datahub.get_return_matrix(
trade_time=datetime(2024, 10, 31, 19, 15),
factors=["forward_ret_raw_1d"],
instrument_ids=["600519.XSHG", "000001.XSHE", "000002.XSHG"],
)
print(result)
print("获取K线")
result = datahub.get_kline(
"5min", instruments=["600519.XSHG", "603350.XSHG"],
start_time=start_time, end_time=end_time, adj_method="backward"
)
print(result)
print("获取指标信息")
result = datahub.get_indicator_info()
print(result)
print("获取指标类型信息")
result = datahub.get_indicator_type()
print(result)
print("获取因子类型信息")
result = datahub.get_factor_type()
print(result)
print("获取行业分类信息")
result = datahub.get_instrument_industry(datetime(2018, 1, 5).date())
print(result)
print("获取交易日")
result = datahub.calendar.get_latest_trade_date(
dt=datetime(2025, 1, 4).date())
print(result)
print("获取主力期货合约信息")
result = datahub.get_future_domain_info("IC")
print(result)
print("获取主力期货快照行情")
result = datahub.get_future_snapshot(
"IC", start_date=start_time.date(), end_date=end_time.date())
print(result)
print("获取return_factor的数据")
df = datahub.get_return_matrix(
trade_time=datetime(2018, 1, 29, 9, 35),
factors=["ret_raw_1d", "ret_raw_3d", "ret_raw_5d"],
instrument_ids=["600519.XSHG", "000001.XSHE", "000002.XSHG"],
)
print(df)
print("获取risk_factor的数据")
df = datahub.get_risk_factor_matrix(
version="rq_v2_sws2021",
trade_time=datetime(2018, 1, 11, 20),
factors=["liquidity", "longterm_reversal", "mid_cap"],
instrument_ids=["600519.XSHG", "000001.XSHE", "000002.XSHG"],
)
print(df)
print("获取最新黑名单列表, 可以指定日期")
df = datahub.get_blacklist(blacklist_ids=["cms_dma_blacklist", "XXX"], end_date=None)
print(df)
print("获取券池列表, 可以指定券池和日期")
df = datahub.get_sbl_list(end_date=datetime(2025, 1, 1))
print(df)
print("获取seq y")
df = datahub.get_seq_y_info(resample_type="time_interval_ms_500",)
print(df)
print("获取seq factor info")
df = datahub.get_seq_factor_info(resample_type="time_interval_ms_500",)
print(df)
print("获取 seq factor")
df = datahub.get_seq_factor_data(
start_time=datetime(2025, 2, 10, 9, 30, 9, 920000),
end_time=datetime(2025, 2, 10, 9, 30, 11, 920000),
factors=["ActiveCashFlowFactors__lookback_tick_30s__time_interval_ms_500__buy_amount"],
instruments=["000029.XSHE"]
)
print(df)
print("获取get_seq_y_stat")
df = datahub.get_seq_y_stat(start_time=datetime(2025, 1, 10, 23), end_time=datetime(2025, 1, 10, 23),
stat_type="std_1d")
print(df)
if __name__ == "__main__":
main()
1 maintainer
Metadata
1 maintainer
Release files for datahub_binary 2.0.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| datahub_binary-2.0.0-cp312-cp312-win_amd64.whl | CPython 3.12 | CPython 3.12 | Windows x86-64 | Details |
Release files / datahub_binary-2.0.0-cp312-cp312-win_amd64.whl
| Download URL | datahub_binary-2.0.0-cp312-cp312-win_amd64.whl |
|---|---|
| Size | 773.0 kB |
| Tags | CPython 3.12 Windows x86-64 |
|
SHA-256 checksum How to use checksums |
93b3b9b70cc62c7f4b8fca4279d2c49756af37058df8b49fb4e06d938c4cb180
|
|
BLAKE2b-256 checksum How to use checksums |
2bdbf0c465e07b67fcef4edb116f0295350071903e92036fdc2ab184526422e9
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
uv/0.11.4 {"installer":{"name":"uv","version":"0.11.4","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":null,"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":null}
|
