一、收益图

二、源代码

# 导入函数库
from jqdata import *
import pandas as pd
import numpy as np
import statsmodels.api as sm

print(f'pd version: {pd.__version__}')  # 0.23.4
print(f'np version: {np.__version__}')  # 1.14.6

# 初始化函数，设定基准等等
def initialize(context):
    # 设定沪深300作为基准
    set_benchmark('000300.XSHG')
    # 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    # 输出内容到日志 log.info()
    log.info('初始函数开始运行且全局只运行一次')

    ### 股票相关设定 ###
    # 股票类每笔交易时的手续费是：买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5块钱
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')

    g.security = '000300.XSHG'
    g.buy_beta, g.sell_beta = 0.7, -0.7
    run_daily(market_open, time='open', reference_security='000300.XSHG')

# 注意：和研究版jupyter稍有区别，因为环境版本不同
def calculate_rsrs(df, n=18):
    """
    计算RSRS右偏标准分指标
    df 包括 high, low
    """
    if len(df) < n:
        raise ValueError("DataFrame长度不足以计算RSRS。")
    if not isinstance(df, pd.DataFrame):
        raise TypeError("df 必须是一个 Pandas DataFrame")
    if 'high' not in df.columns or 'low' not in df.columns:
        raise ValueError("df 必须包含 'high' 和 'low' 列")
    # 1. 计算斜率beta
    beta_list = []
    r2_list = []
    for i in range(n, len(df)):
        # 第一列是常数1（初始值，fit后会调整）；第二列为原本选定的 low 列数值
        x = sm.add_constant(df['low'][i-n:i])
        y = df['high'][i-n:i]
        # y = constant*1 + beta*x  (所以前面加了一列1)
        # OLS 类（普通最小二乘法）
        model = sm.OLS(y, x).fit()
        # 它代表直线的斜率（beta）
        beta = model.params[1]
        # R²:模型拟合度、解释力，值在 0 到 1 之间，越接近 1 表示模型拟合的越好。
        r2 = model.rsquared
        beta_list.append(beta)
        r2_list.append(r2)
    
    df = df[n:].copy()
    assert len(df) == len(beta_list), "长度不匹配：df[n:]与beta_list"
    assert len(df) == len(r2_list), "长度不匹配：df[n:]与r2_list"
    df['beta'] = beta_list
    df['r2'] = r2_list
    
    # 2. 计算标准分 （将斜率转化成符合正态分布的）
    df['beta_mean'] = df['beta'].mean()
    df['beta_std'] = df['beta'].std()
    df['z_score'] = (df['beta'] - df['beta_mean']) / df['beta_std']
    
    # 3. 右偏修正
    df['adj_z'] = df['z_score'] * df['r2']  # 修正标准分；标准分数与模型的解释力（R²）结合
    df['right_skew_z'] = df['adj_z'] * df['beta']  # 右偏标准分；标准分与回归系数结合使用
    return df


## 开盘时运行函数
def market_open(context):
    # 获取股票的价格
    data_df = get_price(g.security, end_date=context.previous_date, count=18+10, frequency='daily', fields=['high','low'])
    # 取得当前的现金
    cash = context.portfolio.available_cash
    rsrs_df = calculate_rsrs(data_df)
    
    if rsrs_df.iloc[-1]['right_skew_z']>g.buy_beta and cash>0:
        # 记录这次买入
        log.info("买入 %s" % (g.security))
        # 用所有 cash 买入股票
        order_value(g.security, cash)
    elif rsrs_df.iloc[-1]['right_skew_z']<g.sell_beta and context.portfolio.positions[g.security].closeable_amount > 0:
        # 记录这次卖出
        log.info("卖出 %s" % (g.security))
        # 卖出所有股票,使这只股票的最终持有量为0
        order_target(g.security, 0)

以上仅为rsrs的基础用法，有更有意思的正确用法，请参见：量化交易 - RSRS（阻力支撑相对强度）- 正确用法-CSDN博客