聚宽和金字塔选股差别很大

106673

我想做一个涨停选股的测试，然后在聚宽中进行回测得到交易明细，选出的股票符合我的要求，然后我在金字塔里想选出4.3那一天的股票进行条件选股的验证，结果发现选出的股票和聚宽里回测出的差异很大，但代码都是一样的逻辑，为什么会有这么大的差别呢？

聚宽策略代码如下：from jqdata import *import numpy as np
import pandas as pd


def initialize(context):
    set_benchmark('000001.XSHG')
    set_option('avoid_future_data', True)
    set_option('use_real_price', True)

    set_slippage(FixedSlippage(0), type='stock')
    set_order_cost(
        OrderCost(
            open_tax=0,
            close_tax=0,
            open_commission=0,
            close_commission=0,
            min_commission=0
        ),
        type='stock'
    )

    g.max_holdings = 20
    g.price_tick = 0.01
    g.batch_size = 800

    g.month_key = None
    g.monthly_buyable = set()

    g.pending_buys = []
    g.pending_sells = {}

    run_daily(build_signals_after_close, time='after_close')
    run_daily(execute_orders_at_open, time='open')


def build_signals_after_close(context):
    signal_date = context.current_dt.date()

    refresh_monthly_buyable(context, signal_date)

    holdings = list(context.portfolio.positions.keys())
    signal_universe = list(g.monthly_buyable.union(set(holdings)))

    if len(signal_universe) == 0:
        clear_pending_orders()
        return

    adjusted_data = load_daily_data(
        signal_universe,
        signal_date,
        119,
        ['close', 'money'],
        'pre'
    )

    raw_data = load_daily_data(
        signal_universe,
        signal_date,
        30,
        ['close', 'high_limit', 'money'],
        None
    )

    adjusted_groups = group_by_code(adjusted_data)
    raw_groups = group_by_code(raw_data)

    g.pending_sells = {}

    holding_st_status = get_holding_st_status(holdings, signal_date)

    for stock in holdings:
        if holding_st_status.get(stock, True):
            g.pending_sells[stock] = 'st_or_risk_warning'
            continue

        adjusted = adjusted_groups.get(stock)

        if is_below_ma5(adjusted):
            g.pending_sells[stock] = 'close_below_ma5'

    candidates = []

    for stock in g.monthly_buyable:
        if stock in holdings:
            continue

        candidate = evaluate_candidate(
            stock,
            adjusted_groups.get(stock),
            raw_groups.get(stock)
        )

        if candidate is not None:
            candidates.append(candidate)

    candidates.sort(
        key=lambda item: (
            -item['limit_up_count'],
            -item['money'],
            item['stock']
        )
    )

    g.pending_buys = candidates

    log.info(
        'signal_date=%s candidates=%d sell_signals=%d'
        % (
            signal_date,
            len(g.pending_buys),
            len(g.pending_sells)
        )
    )


def execute_orders_at_open(context):
    current_data = get_current_data()
    current_holdings = set(context.portfolio.positions.keys())
    submitted_sells = set()

    for stock, reason in g.pending_sells.items():
        if stock not in context.portfolio.positions:
            continue

        position = context.portfolio.positions[stock]
        data = current_data[stock]

        if position.closeable_amount <= 0:
            log.info('sell_skip_t_plus_one=%s' % stock)
            continue

        if data.paused or data.day_open is None or data.day_open <= 0:
            log.info('sell_skip_paused_or_missing_open=%s' % stock)
            continue

        if data.day_open <= data.low_limit + g.price_tick:
            log.info('sell_skip_open_limit_down=%s' % stock)
            continue

        result = order(stock, -position.closeable_amount)

        if result is not None:
            submitted_sells.add(stock)
            log.info('sell_submit=%s reason=%s' % (stock, reason))

    expected_holdings = current_holdings.difference(submitted_sells)
    available_slots = max(0, g.max_holdings - len(expected_holdings))

    if available_slots <= 0:
        clear_pending_orders()
        return

    target_value = context.portfolio.total_value / float(g.max_holdings)
    cash_budget = context.portfolio.available_cash
    bought_count = 0

    for item in g.pending_buys:
        if bought_count >= available_slots:
            break

        stock = item['stock']

        if stock in current_holdings:
            continue

        data = current_data[stock]
        open_price = data.day_open
        signal_close = item['signal_close']

        if data.paused or open_price is None or open_price <= 0:
            log.info('buy_skip_paused_or_missing_open=%s' % stock)
            continue

        if data.is_st or 'ST' in data.name.upper() or '*' in data.name:
            log.info('buy_skip_current_st=%s' % stock)
            continue

        if open_price > signal_close * 1.03:
            log.info('buy_skip_gap_above_3pct=%s' % stock)
            continue

        if open_price >= data.high_limit - g.price_tick:
            log.info('buy_skip_open_limit_up=%s' % stock)
            continue

        order_cash = min(target_value, cash_budget)
        shares = int(order_cash / open_price / 100) * 100

        if shares < 100:
            log.info('buy_skip_insufficient_cash_or_lot=%s' % stock)
            continue

        result = order(stock, shares)

        if result is not None:
            cash_budget -= shares * open_price
            bought_count += 1
            log.info('buy_submit=%s shares=%d' % (stock, shares))

    clear_pending_orders()


def refresh_monthly_buyable(context, signal_date):
    month_key = (signal_date.year, signal_date.month)

    if g.month_key == month_key:
        return

    securities = get_all_securities(['stock'], date=signal_date)
    stocks = list(securities.index)
    st_parts = []

    for start in range(0, len(stocks), g.batch_size):
        batch = stocks[start:start + g.batch_size]

        st_data = get_extras(
            'is_st',
            batch,
            start_date=signal_date,
            end_date=signal_date,
            df=True
        )

        if st_data is not None and len(st_data) > 0:
            st_parts.append(st_data.iloc[-1])

    if len(st_parts) == 0:
        g.monthly_buyable = set()
        g.month_key = month_key
        log.error('monthly_st_snapshot_failed=%s' % signal_date)
        return

    st_series = pd.concat(st_parts)
    buyable = []

    for stock in stocks:
        name = str(securities.loc[stock, 'display_name']).upper()
        is_st = bool(st_series.get(stock, True))

        if is_st:
            continue

        if 'ST' in name or '*' in name or '退' in name:
            continue

        buyable.append(stock)

    g.monthly_buyable = set(buyable)
    g.month_key = month_key

    log.info(
        'monthly_buyable_snapshot=%s buyable=%d'
        % (signal_date, len(g.monthly_buyable))
    )


def get_holding_st_status(holdings, signal_date):
    if len(holdings) == 0:
        return {}

    st_data = get_extras(
        'is_st',
        holdings,
        start_date=signal_date,
        end_date=signal_date,
        df=True
    )

    status = {}

    if st_data is None or len(st_data) == 0:
        for stock in holdings:
            status[stock] = True
        return status

    st_series = st_data.iloc[-1]

    for stock in holdings:
        is_st = bool(st_series.get(stock, True))
        status[stock] = is_st

    return status


def evaluate_candidate(stock, adjusted, raw):
    if adjusted is None or raw is None:
        return None

    valid_adjusted = adjusted[
        (adjusted['money'] > 0) &
        adjusted['close'].notna()
    ]

    if len(valid_adjusted) < 119:
        return None

    if len(raw) < 30:
        return None

    raw = raw.tail(30)
    raw_today = raw.iloc[-1]

    if raw_today['money'] <= 0:
        return None

    adjusted_closes = valid_adjusted['close'].values[-119:]

    ma5_series = rolling_mean(adjusted_closes, 5)
    ma30_series = rolling_mean(adjusted_closes, 30)
    ma60_series = rolling_mean(adjusted_closes, 60)

    if len(ma5_series) < 5:
        return None

    if len(ma30_series) < 30:
        return None

    if len(ma60_series) < 60:
        return None

    ma5 = ma5_series[-1]
    slope5 = linear_slope(ma5_series[-5:])
    slope30 = linear_slope(ma30_series[-30:])
    slope60 = linear_slope(ma60_series[-60:])

    if slope5 <= 0 or slope30 <= 0 or slope60 <= 0:
        return None

    if adjusted_closes[-1] <= ma5:
        return None

    limit_up_flags = (
        raw['high_limit'].notna() &
        (raw['high_limit'] > 0) &
        (raw['money'] > 0) &
        (abs(raw['close'] - raw['high_limit']) <= g.price_tick)
    )

    limit_up_count = int(limit_up_flags.sum())

    if not bool(limit_up_flags.iloc[-1]):
        return None

    if limit_up_count < 2:
        return None

    return {
        'stock': stock,
        'signal_close': float(raw_today['close']),
        'limit_up_count': limit_up_count,
        'money': float(raw_today['money'])
    }


def is_below_ma5(adjusted):
    if adjusted is None or len(adjusted) == 0:
        return False

    valid_adjusted = adjusted[
        (adjusted['money'] > 0) &
        adjusted['close'].notna()
    ]

    if len(valid_adjusted) < 5:
        return False

    closes = valid_adjusted['close'].values
    ma5 = np.mean(closes[-5:])

    return closes[-1] < ma5


def load_daily_data(stocks, end_date, count, fields, fq):
    parts = []

    for start in range(0, len(stocks), g.batch_size):
        batch = stocks[start:start + g.batch_size]

        data = get_price(
            batch,
            end_date=end_date,
            count=count,
            frequency='daily',
            fields=fields,
            skip_paused=False,
            fq=fq,
            panel=False
        )

        if data is not None and len(data) > 0:
            parts.append(data)

    if len(parts) == 0:
        return pd.DataFrame()

    return pd.concat(parts, ignore_index=True)


def group_by_code(data):
    if data is None or len(data) == 0:
        return {}

    time_column = 'time' if 'time' in data.columns else 'date'
    groups = {}

    for stock, frame in data.groupby('code'):
        groups[stock] = frame.sort_values(
            time_column
        ).reset_index(drop=True)

    return groups


def rolling_mean(values, window):
    if len(values) < window:
        return np.array([])

    return np.convolve(
        values,
        np.ones(window) / float(window),
        mode='valid'
    )


def linear_slope(values):
    if len(values) < 2:
        return 0.0

    x = np.arange(len(values), dtype=float)
    return float(np.polyfit(x, values, 1)[0])


def clear_pending_orders():
    g.pending_buys = []
    g.pending_sells = {}


金字塔的代码如下：
// 名称：ThreeMASlope_TwoLimitUp_Pool
// 周期：日线
// 用途：尽量复刻聚宽收盘后的候选股票池

MA5:=MA(CLOSE,5);
MA30:=MA(CLOSE,30);
MA60:=MA(CLOSE,60);

UP5:=SLOPE(MA5,5)>0;
UP30:=SLOPE(MA30,30)>0;
UP60:=SLOPE(MA60,60)>0;

ABOVE5:=CLOSE>MA5;

LIMITUP:=ROUNDS(OCLOSE,2)>=ROUNDS(UPLMTPRICE,2);

TWO_LIMITUP:=COUNT(LIMITUP,30)>=2;

NONST:=INSTRUMENTSING(0)=0 AND INSTRUMENTSING(1)=0;

POOL:
    UP5
    AND UP30
    AND UP60
    AND ABOVE5
    AND LIMITUP
    AND TWO_LIMITUP
    AND NONST;

资深技术05

你可以看下关键的计算是否存在较大的差异比如SLOPE 这种。

PEL的好弄，你直接加载带图上 你可以直接看各个变量的计算结果。py的我建议你针对特定品种输出相关的变量只，和PEL的做对比。
在相同品种上，对比使用的数据量，变量结果等等。  

VIP客服01

只测试中间一天还是日线，有可能会因为日线数据不足导致没法正常工作，建议你补全历史日线后，不要使用这个阶段选股，测试最后一天的选股条件