策略编写

策略编写

Jiuhuang支持自定义策略，只需继承 Strategy 基类并实现 _execute_one 方法。

基本结构

from jiuhuang.strategy import Strategy
import pandas as pd

class MyStrategy(Strategy):
    def __init__(self, entry_window: int = 20, exit_window: int = 10):
        self.entry_window = entry_window
        self.exit_window = exit_window

    def _execute_one(self, data: pd.DataFrame) -> pd.DataFrame:
        """对单个标的生成买卖信号"""
        data = data.copy()

        # 计算滚动最高价和最低价
        data["entry_high"] = data["high"].rolling(window=self.entry_window, min_periods=1).max()
        data["exit_low"] = data["low"].rolling(window=self.exit_window, min_periods=1).min()

        # 生成买卖信号
        data["buy_signal"] = (data["close"] > data["entry_high"].shift(1)).astype(int)
        data["sell_signal"] = (data["close"] < data["exit_low"].shift(1)).astype(int)

        # 清理临时列
        data = data.drop(["entry_high", "exit_low"], axis=1)
        return data

关键要点

1. 继承 Strategy 基类：自定义策略必须继承 Strategy
2. 实现 _execute_one 方法：必须实现此方法
3. 输入参数：入参为 pandas.DataFrame，包含单只股票的历史数据
4. 输出要求：出参需包含 buy_signal 和 sell_signal 两列
   • buy_signal: 买入信号（1 表示买入，0 表示不买入）
   • sell_signal: 卖出信号（1 表示卖出，0 表示不卖出）

使用自定义策略

from jiuhuang.data import JiuhuangData, DataTypes
from jiuhuang.strategy import Strategy
from jiuhuang.backtest import backtest
import pandas as pd

class MyStrategy(Strategy):
    def __init__(self, entry_window: int = 20, exit_window: int = 10):
        self.entry_window = entry_window
        self.exit_window = exit_window

    def _execute_one(self, data: pd.DataFrame) -> pd.DataFrame:
        data = data.copy()
        data["entry_high"] = data["high"].rolling(window=self.entry_window, min_periods=1).max()
        data["exit_low"] = data["low"].rolling(window=self.exit_window, min_periods=1).min()
        data["buy_signal"] = (data["close"] > data["entry_high"].shift(1)).astype(int)
        data["sell_signal"] = (data["close"] < data["exit_low"].shift(1)).astype(int)
        data = data.drop(["entry_high", "exit_low"], axis=1)
        return data

# 定义策略
strategies = {
    "我的海龟策略": MyStrategy(entry_window=20, exit_window=10),
}

# 获取数据
jh = JiuhuangData()
symbols = ["000001", "600036", "600519"]
stock_price = jh.get_data(
    DataTypes.STOCK_ZH_A_HIST_QFQ,
    start="2024-12-25",
    end="2026-03-11",
    symbol=",".join(symbols),
)
stock_info = jh.get_data(DataTypes.STOCK_INDIVIDUAL_INFO_EM)

# 执行回测
trading_history, backtest_perf = backtest(strategies, stock_price, stock_info)

多进程并行

jiuhuang 会默认使用多进程并行进行回测，所以速度很快。

下一步

• 内置策略 - 使用内置策略