Qlib 中文文档

pip install numpy
pip install --upgrade cython

git clone https://github.com/microsoft/qlib.git && cd qlib
python setup.py install

python scripts/get_data.py qlib_data --target_dir ~/.qlib/qlib_data/cn_data --region cn

cd examples  # 避免在包含 `qlib` 的目录下运行
qrun benchmarks/LightGBM/workflow_config_lightgbm.yaml

pip install pyqlib

$ pip install numpy
$ pip install --upgrade cython
$ git clone https://github.com/microsoft/qlib.git && cd qlib
$ python setup.py install

>>> import qlib
>>> qlib.__version__
'0.9.8.dev11'

python scripts/get_data.py qlib_data --target_dir ~/.qlib/qlib_data/cn_data --region cn

import qlib
# region 在 [REG_CN, REG_US] 中选择
from qlib.constant import REG_CN

provider_uri = "~/.qlib/qlib_data/cn_data"  # 数据目录
qlib.init(provider_uri=provider_uri, region=REG_CN)

qlib.init(provider_uri=provider_uri, region=REG_CN, exp_manager={
    "class": "MLflowExpManager",
    "module_path": "qlib.workflow.expm",
    "kwargs": {
        "uri": "python_execution_path/mlruns",
        "default_exp_name": "Experiment",
    }
})

qlib.init(provider_uri=provider_uri, region=REG_CN, mongo={
    "task_url": "mongodb://localhost:27017/",  # mongo URI
    "task_db_name": "rolling_db",  # 任务管理的数据库名称
})

>>> import qlib
>>> qlib.init(provider_uri='~/.qlib/qlib_data/cn_data')

>>> from qlib.data import D
>>> D.calendar(start_time='2010-01-01', end_time='2017-12-31', freq='day')[:2]
[Timestamp('2010-01-04 00:00:00'), Timestamp('2010-01-05 00:00:00')]

>>> from qlib.data import D
>>> instruments = D.instruments(market='csi300')
>>> D.list_instruments(instruments=instruments, start_time='2010-01-01',
                       end_time='2017-12-31', as_list=True)[:6]
['SH600036', 'SH600110', 'SH600087', 'SH600900', 'SH600089', 'SZ000912']

>>> from qlib.data import D
>>> instruments = ['SH600000']
>>> fields = ['$close', '$volume', 'Ref($close, 1)', 'Mean($close, 3)', '$high-$low']
>>> D.features(instruments, fields, start_time='2010-01-01',
               end_time='2017-12-31', freq='day').head()

>>> from qlib.data.filter import NameDFilter
>>> nameDFilter = NameDFilter(name_rule_re='SH[0-9]{4}55')
>>> instruments = D.instruments(market='csi300', filter_pipe=[nameDFilter])
>>> D.list_instruments(instruments=instruments, start_time='2015-01-01',
                       end_time='2016-02-15', as_list=True)
['SH600655', 'SH601555']

>>> from qlib.data.filter import ExpressionDFilter
>>> expressionDFilter = ExpressionDFilter(rule_expression='$close>2000')
>>> instruments = D.instruments(market='csi300', filter_pipe=[expressionDFilter])

>>> from qlib.data.ops import *
>>> f1 = Feature("high") / Feature("close")
>>> f2 = Feature("open") / Feature("close")
>>> f3 = f1 + f2
>>> f4 = f3 * f3 / f3
>>> data = D.features(["sh600519"], [f4], start_time="20200101")

workflow:
  class: "qlib.workflow.workflow.TemporalWorkflow"
  module_path: "qlib.workflow.workflow"
  kwargs:
    start_time: "2010-01-01"
    end_time: "2017-12-31"
    fit_start_time: "2010-01-01"
    fit_end_time: "2016-12-31"

executor:
  class: "qlib.backtest.executor.BacktestExecutor"
  module_path: "qlib.backtest.executor"
  kwargs:
    trade_exchange:
      class: "qlib.backtest.exchange.SimpleExchange"
      module_path: "qlib.backtest.exchange"
      kwargs:
        limit_threshold: 0.2

task:
  class: "qlib.workflow.task_management.TrainingTask"
  module_path: "qlib.workflow.task_management"
  kwargs:
    dataset:
      class: "qlib.data.dataset.DatasetH"
      module_path: "qlib.data.dataset"
      kwargs:
        segments:
          train:
            start_time: "2010-01-01"
            end_time: "2015-12-31"
          type: "train"
    model:
      class: "qlib.contrib.model.gbdt.LGBModel"
      module_path: "qlib.contrib.model.gbdt"
      kwargs:
        loss: "mse"
        col: "mse"

qrun workflow_config.yaml

from qlib.contrib.model.gbdt import LGBModel

model = LGBModel(
    loss="mse",
    col="mse",
)
model.fit(dataset)
pred = model.predict(dataset)

from qlib.contrib.model.pytorch_nn import DNNModelPytorch

model = DNNModelPytorch(
    dnn_layers=(64, 32,),
    dropout=0.2,
)
model.fit(dataset)
pred = model.predict(dataset)

from qlib.contrib.model.pytorch_lstm import LSTMModel

model = LSTMModel(
    dnn_layers=(64,),
    lstm_layers=(32,),
    dropout=0.2,
)
model.fit(dataset)
pred = model.predict(dataset)

# 推荐的workflow文件组织结构
project/
├── configs/
│   ├── base_config.yaml       # 基础配置（数据、市场）
│   ├── model_lgbm.yaml        # 模型配置
│   ├── model_mlp.yaml         # 模型配置
│   └── strategy_topk.yaml     # 策略配置
├── scripts/
│   ├── run_exp.py             # 实验运行脚本
│   └── analyze_results.py     # 结果分析脚本
└── notebooks/
    └── visualize.ipynb        # 可视化分析

from qlib.workflow import R
from qlib.tests.config import TEST_CONFIG

models = ['LGBModel', 'MLPModel', 'LSTMModel']
results = {}

with R.start(experiment_name="model_comparison"):
    for model_name in models:
        # 运行每个模型
        recorder = R.get_recorder()
        results[model_name] = recorder.list_metrics()

# 对比结果
for model, metrics in results.items():
    print(f"{model}: IC={metrics['IC']}, RankIC={metrics['Rank IC']}")

# LightGBM并行化加速
model_config = {
    "class": "LGBModel",
    "module_path": "qlib.contrib.model.gbdt",
    "kwargs": {
        "num_threads": 20,      # 利用多核
        "learning_rate": 0.02,    # 降低学习率提高精度
        "num_leaves": 210,      # 增加叶节点数提升拟合能力
        "feature_fraction": 0.9,  # 特征采样防止过拟合
        "bagging_fraction": 0.8, # 数据采样
        "early_stopping_rounds": 50  # 早停
    }
}

# PyTorch GPU加速
mlp_config = {
    "class": "DNNModelPytorch",
    "module_path": "qlib.contrib.model.pytorch_nn",
    "kwargs": {
        "GPU": 0,              # 使用GPU 0
        "batch_size": 2048,     # 增大批次提高GPU利用率
        "lr": 0.002,
        "max_steps": 10000
    }
}

from qlib.contrib.strategy.topk import TopkStrategy

strategy = TopkStrategy(
    topk=30,
    n_drop=5,
    method="weight",
)

from qlib.contrib.strategy.rl_strategy import TwinFilterStrategy

strategy = TwinFilterStrategy(
    transaction=0.0015,
    trade_exchange=exchange,
)

from qlib.contrib.strategy.strategy import TopkDropoutStrategy
import pandas as pd
import numpy as np

class RiskControlledStrategy(TopkDropoutStrategy):
    """
    带风险控制的Topk策略
    - 限制单股权重
    - 控制行业集中度
    - 过滤低流动性股票
    """
    
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.max_weight_per_stock = 0.05  # 单股最大5%
        self.max_industry_concentration = 0.3  # 行业最大30%
    
    def generate_order_dir(self, pred_score, **kwargs):
        # 调用父类方法生成初始仓位
        W = super().generate_order_dir(pred_score, **kwargs)
        
        # 应用风险约束
        W = self._apply_weight_cap(W)
        W = self._apply_industry_cap(W)
        
        # 重新归一化
        W = W / W.sum()
        
        return W
    
    def _apply_weight_cap(self, W):
        # 限制单股权重
        W = W.clip(upper=self.max_weight_per_stock)
        return W
    
    def _apply_industry_cap(self, W):
        # 简化版本：假设已有行业分类数据
        # 实际需要加载行业数据并计算集中度
        return W

# 使用自定义策略
strategy_config = {
    "class": "RiskControlledStrategy",
    "module_path": "your_module",
    "kwargs": {
        "topk": 50,
        "n_drop": 5
    }
}

# 自定义Alpha因子集合
from qlib.data.dataset.handler import DataHandlerLP

# 定义自定义因子
custom_features = [
    # 动量类
    "Ref($close, 20) / $close - 1",           # 20日收益率
    "Ref($close, 60) / $close - 1",           # 60日收益率
    "($close - Ref($close, 5)) / Ref($close, 5)",  # 5日收益率
    
    # 反转类
    "-Ref($close, 3) / $close + 1",           # 3日反转
    "-($high - $low) / $close",               # 日内振幅
    
    # 波动类
    "Std($close, 20) / Mean($close, 20)",     # 20日波动率
    "Std($volume, 10) / Mean($volume, 10)",   # 10日成交量波动
    
    # 成交量相关
    "$volume / Mean($volume, 20)",            # 相对成交量
    "($volume * $close) / Mean($volume * $close, 20)", # 相对成交额
    
    # 技术指标
    "($close - Mean($close, 20)) / Std($close, 20)", # Z-score
    "($close - Mean($close, 10)) / Mean($close, 10)", # 偏离度
    
    # 跨截面特征
    "Rank($close / Ref($close, 1))",          # 当日收益率排名
    "Rank($volume)",                           # 成交量排名
]

# 数据处理器配置
handler_config = {
    "class": "DataHandlerLP",
    "module_path": "qlib.data.dataset.handler",
    "kwargs": {
        "start_time": "2008-01-01",
        "end_time": "2020-08-01",
        "fit_start_time": "2008-01-01",
        "fit_end_time": "2014-12-31",
        "instruments": "csi300",
        "infer_processors": [
            {"class": "RobustZScoreNorm", "kwargs": {"fields_group": "feature", "clip_outlier": True}},
            {"class": "Fillna", "kwargs": {"fields_group": "feature"}},
        ],
        "learn_processors": [
            {"class": "DropnaLabel"},
            {"class": "CSRankNorm", "kwargs": {"fields_group": "label"}},
        ],
    }
}

# 标签定义：预测未来2天收益率
label = ["Ref($close, -2) / Ref($close, -1) - 1"]

# 示例：复杂表达式
fields = [
    "Ref($close, 1) / $close - 1",  # 1日收益率
    "Mean($close, 5) / $close - 1",  # 5日均值收益率
    "Std($close, 10) / Mean($close, 10)",  # 变异系数
    "Rank($close) / len($close)",  # 价格排名
    "$close > Ref($close, 1)",  # 是否上涨
]

# 集成自定义数据源（如Tushare、AKShare）
import pandas as pd
from qlib.data import D

# 方案1：使用DataProvider接口
from qlib.data.base import BaseProvider

class CustomDataProvider(BaseProvider):
    def __init__(self, api_token):
        self.api_token = api_token
    
    def get_data(self, instruments, start_time, end_time):
        # 调用外部API获取数据
        df = self.fetch_from_api(instruments, start_time, end_time)
        # 转换为Qlib格式
        return self.to_qlib_format(df)

# 方案2：直接转换为Qlib格式
def convert_to_qlib(source_df):
    """
    将标准DataFrame转换为Qlib格式
    
    Parameters
    ----------
    source_df : pd.DataFrame
        columns = [datetime, instrument, open, high, low, close, volume]
    
    Returns
    -------
    pd.DataFrame
        Qlib标准格式
    """
    df = source_df.copy()
    
    # 重命名列
    df.columns = ['date', 'instrument', 'open', 'high', 
                  'low', 'close', 'volume']
    
    # 设置索引
    df = df.set_index(['date', 'instrument'])
    
    # 保存为Qlib格式
    df.to_pickle('custom_data.bin')
    
    return df

# 使用自定义数据
qlib.init(
    provider_uri='custom_data/',
    region=REG_CN,
    custom_data_provider=CustomDataProvider(api_token='your_token')
)

# 动量因子
alpha1 = "($close - Ref($close, 20)) / Std($close, 20)"

# 反转因子
alpha2 = "-Rank($close) / len($close)"

# 波动因子
alpha3 = "($high - $low) / Mean($close, 10)"

from qlib.contrib.data.alpha import get_all_alpha

# 获取所有 Alpha 表达式
alpha_dict = get_all_alpha()

# 使用 Alpha
alpha_expr = alpha_dict['alpha036']

from qlib.contrib.evaluate import backtest
from qlib.contrib.strategy.rl_strategy import TwinFilterStrategy

# 配置 RL 工作流
config = {
    "env": {
        "class": "qlib.contribute.simulator.ExchangeSimulator",
        "module_path": "qlib.contribute.simulator",
    },
    "agent": {
        "class": "qlib.rl.agent.Agent",
        "module_path": "qlib.rl.agent",
    },
}

# 运行 RL 训练
portfolio_result, indicator_result = backtest(strategy, exchange)

# PPO订单执行策略（简化版）
from qlib.rl import RLOrderExecPolicy
from qlib.contrib.strategy import TopkDropoutStrategy

# 配置RL策略
rl_config = {
    "class": "RLOrderExecPolicy",
    "module_path": "qlib.rl.order_execution",
    "kwargs": {
        "observation_space": 30,      # 状态维度（价格、成交量、订单簿等）
        "action_space": 10,          # 动作空间（10个离散动作）
        "freq": "1min",                # 1分钟级别决策
        "total_steps": 10000,          # 训练步数
        "reward_scaler": 1e4,           # 奖励缩放因子
    }
}

# 使用RL策略作为执行器
executor_config = {
    "class": "NestedExecutor",
    "kwargs": {
        "outer_strategy": TopkDropoutStrategy,
        "inner_executor": rl_config
    }
}

# 关键要点：
# 1. 外层策略决定买哪些股票（日频）
# 2. 内层RL策略决定如何拆单执行（分钟级）
# 3. 奖励函数考虑滑点、成交速度、交易成本

# 训练完成后保存模型
model.save("my_model.pkl")

# 加载已保存的模型
model.load("my_model.pkl")

# 获取最新数据
latest_data = D.features(instruments, fields)

# 执行预测
predictions = model.predict(latest_data)

# 生产环境架构
┌─────────────────────────────────────────────────────┐
│                   Data Layer                         │
│  ├─ Market Data (实时行情)                           │
│  ├─ Fundamental Data (财务数据)                      │
│  └─ Alternative Data (替代数据)                      │
└──────────────────┬──────────────────────────────────┘
                   │
        ┌──────────▼──────────┐
        │  Feature Engine     │ (计算特征)
        │  - 特征计算         │
        │  - 数据清洗         │
        │  - 特征缓存         │
        └──────────┬──────────┘
                   │
        ┌──────────▼──────────┐
        │  Model Service      │ (模型推理)
        │  - 模型加载         │
        │  - 批量预测         │
        │  - 结果缓存         │
        └──────────┬──────────┘
                   │
        ┌──────────▼──────────┐
        │  Signal Generator   │ (生成信号)
        │  - 风险控制         │
        │  - 组合优化         │
        │  - 订单生成         │
        └──────────┬──────────┘
                   │
        ┌──────────▼──────────┐
        │  Execution Engine   │ (执行交易)
        │  - 订单路由         │
        │  - 滑点控制         │
        │  - 仓位管理         │
        └─────────────────────┘

import qlib
from qlib.workflow.expm import MLflowExpManager

# 初始化 Qlib 并配置 MLflow
qlib.init(
    provider_uri=provider_uri,
    region=REG_CN,
    exp_manager={
        "class": "MLflowExpManager",
        "module_path": "qlib.workflow.expm",
        kwargs: {
            "uri": "mlruns",  # MLflow 跟径
            "default_exp_name": "qlib_experiment",
        }
    }
)

# 训练完成后，指标会自动记录到 MLflow
# 也可以手动记录
with mlflow.start_run():
    mlflow.log_metric("sharpe", 1.5)
    mlflow.log_metric("max_drawdown", -0.08)

from qlib.contrib.model.pytorch_nn import DNNModelPytorch
from qlib.contrib.model.base import Model
from qlib.data.dataset import DatasetH

class CustomModel(DNNModelPytorch):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
    
    def fit(self, dataset: DatasetH):
        # 自定义训练逻辑
        super().fit(dataset)
    
    def predict(self, dataset: DatasetH):
        # 自定义预测逻辑
        return super().predict(dataset)

model:
  class: "path.to.CustomModel"
  module_path: "path.to.module"
  kwargs:
    dnn_layers: [128, 64, 32]
    dropout: 0.2

# 导入订单簿数据到Qlib
cd examples/orderbook_data

# 初始化数据库（Arctic或MongoDB）
python create_dataset.py initialize_library

# 导入数据
python create_dataset.py import_data \
    --source_dir /path/to/lob_data \
    --freq 1min \
    --backend arctic

# 数据格式要求：
# - datetime: timestamp
# - instrument: 股票代码
# - bid/ask: 买卖价格
# - bid_vol/ask_vol: 买卖量

# 高频特征工程（订单簿数据）
from qlib.data import D

# 1. 买卖价差（Spread）
spread = "($ask - $bid) / (($ask + $bid) / 2)"

# 2. 订单簿失衡（Imbalance）
imbalance = "($bid_vol - $ask_vol) / ($bid_vol + $ask_vol)"

# 3. 中点价格变动
midpoint_move = "Ref(($bid + $ask) / 2, 1) / Ref(($bid + $ask) / 2, 0) - 1"

# 4. 加权平均价格（VWAP）
vwap = "Mean(($bid * $ask_vol + $ask * $bid_vol) / ($bid_vol + $ask_vol), 60)"

# 5. 波动率（Realized Volatility）
realized_vol = "Std(Ref($close, 1) / Ref($close, 0) - 1, 60)"

features = [spread, imbalance, midpoint_move, vwap, realized_vol]
df = D.features(instruments='csi300', fields=features, freq='1min')

import qlib.contrib.report as qcr

# 查看所有支持的报告
print(qcr.GRAPH_NAME_LIST)

# 生成投资组合报告
qcr.analysis_position.report_graph(
    pred=prediction,
    return_array=returns,
    **config
)

# 嵌套决策执行示例
from qlib.contrib.strategy.hf_trading import NestedStrategy

# 配置多层决策
strategy = NestedStrategy(
    outer_freq='day',
    inner_freq='5min',
)

from qlib.workflow.task.gen import RollingGen

# 生成滚动窗口任务
task_gen = RollingGen(
    step=60,  # 60天滚动
    task_template=task_config
)
tasks = task_gen.generate(task_config)

from qlib.workflow.task.manage import TaskManager

# 配置 MongoDB
task_manager = TaskManager(task_pool='my_tasks')

# 插入任务
task_manager.insert_task(task_def)

XXX_q.data   # 季度数据
XXX_a.data   # 年度数据
XXX_q.index  # 索引文件（加速查询）

# 参考 scripts/data_collector/pit/README.md
python scripts/data_collector/pit/crawler.py
python scripts/data_collector/pit/converter.py

def predict(self, dataset):
    """
    预测数据集。

    Parameters
    ----------
    dataset : Dataset
        要预测的数据集

    Returns
    -------
    prediction : pd.DataFrame
        预测结果
    """
    pass

pip install black
python -m black . -l 120

# 忽略特定错误
return -ICLoss()(pred, target, index)  # pylint: disable=E1130

flake8 --ignore E501,F541,E402,E741 qlib

pip install -e .[dev]
pre-commit install

pip install -e ".[dev]"

from qlib.model.meta.task import MetaTask

# 创建元任务
meta_task = MetaTask(
    task_list=task_config_list,
    dataset=meta_dataset
)

python -m qlib.server.server

from qlib.config import C

C["server_config"] = {
    "host": "localhost",
    "port": 9999,
}

qlib.init(...)

from qlib.utils.serial import Serializable

class MyModel(Serializable):
    def __init__(self):
        super().__init__()
        self.model = None

# 保存模型
model.save("my_model.pkl")

# 加载模型
model.load("my_model.pkl")

# 下载社区维护的A股数据集
wget https://github.com/chenditc/investment_data/releases/latest/download/qlib_bin.tar.gz
mkdir -p ~/.qlib/qlib_data/cn_data
tar -zxvf qlib_bin.tar.gz -C ~/.qlib/qlib_data/cn_data

# 检查数据健康度
python scripts/check_data_health.py check-data \
    --qlib_dir ~/.qlib/qlib_data/cn_data \
    --missing_threshold 300 \
    --price_step 0.5

RuntimeError: An attempt has been made to start a new process 
before the current process has finished its bootstrapping phase.

import qlib
from qlib.data import D

if __name__ == "__main__":
    qlib.init()
    df = D.features(instruments, fields, ...)

QlibCacheException: redis lock key exists.

$ redis-cli
> select 1
> flushdb

pip install -U python-socketio==5.3.0 python-engineio==3.13.2