量化交易初尝试

本文所有内容均为个人见解。仅表示自己在当前认知情况下自认为正确的观点所作出的行为，如有异议还请谅解与提出交流。头图来自 https://invest101.com.hk/grid-trading

心路历程

自己也算是投资市场多少年的韭菜了，虽然玩的都是小钱，最后统计下来还是亏多赢少。以前读了点小书，看了点小文章，立马自信心爆棚，觉得其他交易者都是定力不行，自己不会被市场的情绪左右。交易过程天天说定投，相信时间的力量，投到后面资金不够，也不知道何时止盈止损，看着赚了以为自己牛逼，看着跌了就闭嘴装死。当然我这里不是指一些会发出信号的主动型基金，而是纯粹自己自我感知良好的操作，现在回首看过去，没有明确的系统与目的，进入投资市场无异于赌博。

近几个月决定自己做一点量化的尝试，毕竟一个东西门槛越高，越少的人会去尝试。赛道越少，竞争越少。虽然大的机构也会有专门的量化投资部门，但毕竟会顾虑到资金和面向市场等多方面因素，和个人投资者要考虑的东西完全不一样。况且人家也不会分享给你他们的策略。投资要趁早，既然自己有一定的能力与基础，早点研究就当是玩玩也是有趣的。

量化过程，简化来说大体可以分为如下几部分：数据收集，策略构建和策略回测。数据收集就是选择合适的平台，能够持续稳定准确的给你提供想要研究市场的数据。策略构建是选择合适的策略或是通过机器学习/深度学习训练得到策略。不同的策略同样有着众多的超参数，构建一个合适的模型同样需要考虑其准确性和泛化性。回测过程考虑最终的指标，可能是夏普比率，又可能仅是利润率，具体情况具体分析。本质上和众多科研流程类似，有科学研究内味了。

对于一个策略来说，核心关键点就是信号（何时交易）和数量。在什么时间买入或卖出，交易的过程具体数量是多少。一步步拆解下来就是：构建好自己的框架，模拟这个过程，选择好合适的品种，同时保证准确性和泛化性。

有些人相信价值投资，进行基本面的研究，但是我感觉自己的认知和眼光还是不够，况且我也拿不到单一公司真正内部的资料，又何来了解一说。现阶段不如交给机器去选择，少一点自己的主观能动性吧。

网格交易策略

经过一段时间的探索，掌握了一些皮毛，目前仍然有许多知识盲点与专业性的东西需要慢慢领会。这里先从简单的入手复现一个以前使用过的网格交易策略。

网格交易主要是在震荡环境中非常有效的策略。随着设定合适的买入卖出网格区间，在区间中做到低价买入，高价卖出，赚一点点中间的波动价格差。

在进行网格交易中有以下关键的点：

  1. 选定合适的品种。例如指数类 ETF（沪深 300，中证 500，兴全趋势混合，创业板指），财务稳定的（银行，保险），可以长期持有的股票等等。
  2. 选定好合适的价格区间。等差区间，等比区间，或是其他更多的策略。
  3. 做好压力测试，提前明确定好上下区间进行止盈止损。以及超出区间后如何进行后续操作。

策略编写

这里选择之前实战中使用过的万马股份 (SZSE:002276) 作为研究对象。后续会对更多的品种进行测试。

回测框架经过多方面对比，我选择了现在在持续维护且运行效率较高的 vectorbt 的付费版本 VectorBT PRO。当然也可以选择一些免费的框架如 vnpy。

如下代码在jupyter notebook中运行，暂不考虑优化效率和面向对象编程，仅初步展示和参考。
导入相关库

import vectorbtpro as vbt  
import numpy as np  
import pandas as pd

从 tradingview 抓取相关数据，以天为单位的K线

list_name = vbt.TVData.list_symbols(symbol_pattern="002276")  
data = vbt.TVData.pull("SZSE:002276")
data.stats()

输出
Start Index            2009-07-10 01:30:00+00:00
End Index              2024-11-13 01:30:00+00:00
Total Duration                3623 days 00:00:00
Total Symbols                                  1
Null Counts: 002276                            0
Name: agg_stats, dtype: object

选取特定时间段并可视化K线

close = data.get('Close')
data=data.loc['2018-01-01':]
data.plot().show()
image.png

定义一个等差网格区间

# 等差计算 输入上下区间，目标投入金额，计划网格数 得到每格买入价格间隔，每格买入数量  
def arithmetic_grid(price_low, price_high, initial_fund, grid_num):  
    # 计算每格价格的差价  
    grid_price_interval = round((price_high - price_low) / (grid_num-1),4)  
    # 等差数列计算公式  
    total_multiplier = grid_num * (price_low + price_high) / 2  
    # 计算每格的交易数量，使得总和符合初始资金  
    share_number = int(initial_fund / total_multiplier)  
    print(f"每格交易价格间隔: {grid_price_interval}，每格交易数量: {share_number}")  
    # 交易间隔  
    grid_interval = [round(price_low + i * grid_price_interval, 4) for i in range(grid_num)]  
    return grid_price_interval,share_number,grid_interval

设定网格最低价价为 6，最高价格 12，总计划投入金额 10000，网格数量 10

grid_price, grid_amount, grid_interval = arithmetic_grid(price_low=6, price_high=12, initial_fund=10000, grid_num=10)  
grid_interval

输出：
每格交易价格间隔: 0.6667，每格交易数量: 111
[6.0,
 6.6667,
 7.3334,
 8.0001,
 8.6668,
 9.3335,
 10.0002,
 10.6669,
 11.3336,
 12.0003]

Gridv1

定义最基本简单的网格交易策略：采用每次交易相同数额，输出 direction 表示买卖信号，输出 grid_interval 表示网格间隔，仅用于可视化。

## 针对股票买入特定份数  
## 每一次买入卖出都会重置基准单位  
  
# 策略在执行之前先买入一定比例，后续使用网格进行交易测算。如果基金呈现上涨趋势，建仓比例较大可以吃到更多利润  
# init_position 初始建仓份数  
# init_price 建仓时价格  
# position_size 已建仓的金额  
# remain_cash 用于执行网格的金额  
# units 持有的仓位  
# base_unit_price 基准买入价格  
# today_unit_price 当日价格  
  
def stock_grid_line_v1(close, remain_cash, init_position,price_low=6, price_high=12):  
      
    init_price = close.iloc[0]  
    units = init_position    
      
    base_unit_price = close.iloc[0]  
    direction = pd.Series(np.full(data.shape[0], 0), index=data.index)  
    # size = pd.Series(np.full(data.shape[0], 0), index=data.index)  
    # 计算网格大小和每格交易数量，采用等差方式  
    grid_price, grid_amount,grid_interval = arithmetic_grid(price_low=price_low, price_high=price_high, initial_fund=remain_cash, grid_num=10)  
    print(f"初始买入价格{init_price}")  
      
    for i in range(1, close.shape[0]):  
        today_unit_price = close.iloc[i]  
          
        # 超出交易区间不做交易  
        if today_unit_price<price_low or today_unit_price > price_high:  
            continue  
        ## sold operation  
        # 当日价格高于网格价，且手里有持仓有剩余  
        if today_unit_price >= base_unit_price+grid_price and units >= grid_amount:  
            units -= grid_amount  
            remain_cash += grid_amount*today_unit_price  
            direction.iloc[i] = -1  
            #重置基准单位  
            base_unit_price = today_unit_price  
              
        ## buy operation  
        # 当日价格低于网格价，且手里有现金  
        if today_unit_price <= base_unit_price-grid_price and remain_cash >= grid_amount*today_unit_price:  
            units += grid_amount  
            remain_cash -= grid_amount*today_unit_price  
            direction.iloc[i] = 1  
            #重置基准单位  
            base_unit_price = today_unit_price  
  
                         
    return direction,grid_interval

设定总计划最多投入金额 10000，初始建仓 400 股

direction,grid_interval = stock_grid_line_v1(close,10000,400)

输出：
每格交易价格间隔: 0.6667，每格交易数量: 111
初始买入价格8.7

可视化

def plot_entry_exit(close, entries, exits, basic_line, grid_interval):  
    fig = close.vbt.plot()  
    entries.vbt.signals.plot_as_entries(close, fig=fig)  
    exits.vbt.signals.plot_as_exits(close, fig=fig)  
    fig.add_shape(  
        type="line",  
        x0=0, x1=1,  # 横坐标范围，0到1表示整个x轴的宽度  
        y0=basic_line, y1=basic_line,  # y0=y1=7 表示水平线位置  
        xref='paper',  # 使用 'paper' 参考系确保线条延伸到整个图宽  
        line=dict(color="Red", width=2, dash="dash")  # 设置线条颜色、宽度和样式  
    )  
    for interval in grid_interval:  
        fig.add_shape(  
        type="line",  
        x0=0, x1=1,  # 横坐标范围，0到1表示整个x轴的宽度  
        y0=interval, y1=interval,  # y0=y1=7 表示水平线位置  
        xref='paper',  # 使用 'paper' 参考系确保线条延伸到整个图宽  
        line=dict(color="Blue", width=2, dash="dash")  # 设置线条颜色、宽度和样式  
    )  
    print(f'买入点数量{entries.vbt.signals.total()},卖出点数量{exits.vbt.signals.total()}')  
    return fig.show()

由于时间有一定滞后性，默认为得到交易信号后一天才开始交易。

# 实际操作买入卖出时间向后推迟一个单位
entries = (direction == 1).vbt.signals.fshift() 
exits = (direction == -1).vbt.signals.fshift()
plot_entry_exit(close,entries,exits,basic_line=close.iloc[0],grid_interval=grid_interval)
image.png

进行实际回测，这里主要依赖vectorbt自带的回测方法，直接调用即可。忽略手续费。

pf = vbt.Portfolio.from_signals(  
    close=close,   
    entries=entries,   
    exits=exits,   
    init_cash=10000, # 初始资金  
    init_position= 400, #初始仓位  
    init_price = close.iloc[0], #初始建仓金额  
    size=111, # 使用amount类型，每次进行固定股数的交易  
    size_type="amount",  
    # fees=0.001,  
    accumulate=True,  
)
pf.stats()
image.png

输出解析

  • Total Return 最终投资利润增长
  • Benchmark Return 基线收益（一直持有收益）
  • Total Time Exposure 总暴露在市场中的时间
  • Max Gross Exposure 最大交易中使用的现金量
  • Max Drawdown 最大回撤，即组合价值在创下新高后下跌的最大距离
  • Total Orders 总订单次数
  • Total Trades 总交易次数。Vectorbt 中仅卖出操作计作交易。一旦卖出订单被执行（通过减少或关闭 position），就会根据加权平均进入和退出价格计算损益 (the profit and loss, PnL)
  • Win Rate 交易（卖单）产生利润的概率
  • Best Trade 最好交易的利润率
  • Worst Trade 最差交易的利润率
  • Profit Factor 利润因子
  • Expectancy 预期 average profitability 平均盈利能力

回报率 25%，还是可以的。

Gridv2

简单改进一下网格策略：每次交易网格后，下一个网格按照一定比例增加份额。输出多增加一个 size ，表示交易的数量

## 针对股票买入特定份数  
## 每一次买入卖出都会重置基准单位  
## 大网格增加份额。  
  
# 策略在执行之前先买入一定比例，后续使用网格进行交易测算。
# init_position 初始建仓份数  
# init_price 建仓时价格  
# position_size 已建仓的金额  
# remain_cash 用于执行网格的金额  
# units 持有的仓位  
# base_unit_price 基准买入价格  
# today_unit_price 当日价格  
  
def stock_grid_line_v2(close, remain_cash, init_position,price_low=6, price_high=12):  
    init_price = close.iloc[0]  
    units = init_position  
  
    base_unit_price = close.iloc[0]  
    direction = pd.Series(np.full(data.shape[0], 0), index=data.index)  
    size = pd.Series(np.full(data.shape[0], 0), index=data.index)  
  
    # 计算网格大小和每格交易数量，采用等差方式  
    grid_price, grid_amount, grid_interval = arithmetic_grid(price_low=price_low, price_high=price_high, initial_fund=remain_cash,  
                                                             grid_num=10)  
    print(f"初始买入价格{init_price}")  
  
    # 记录操作位置  
    position_grid = 0  
    # 增加网格份额   
    grid_amount_increase = int(grid_amount * 0.02)  
    print(f"网格增加数量{grid_amount_increase}")  
  
    for i in range(1, close.shape[0]):  
        today_unit_price = close.iloc[i]  
        buy_grid_amount = grid_amount + grid_amount_increase * position_grid  
        sell_grid_amount = grid_amount + grid_amount_increase * position_grid  
          
        # 超出交易区间不做交易  
        if today_unit_price<price_low or today_unit_price > price_high:  
            continue  
  
        ## sold operation  
        # 当日价格高于网格价，且手里有持仓有剩余  
        if today_unit_price >= base_unit_price + grid_price and units >= sell_grid_amount:  
            units -= sell_grid_amount  
            remain_cash += sell_grid_amount * today_unit_price  
            direction.iloc[i] = -1  
            size.iloc[i] = sell_grid_amount  
  
            # 重置调整基准  
            base_unit_price = today_unit_price  
            position_grid -= 1  
  
        ## buy operation  
        # 当日价格低于网格价，且手里有现金  
        if today_unit_price <= base_unit_price - grid_price and remain_cash >= buy_grid_amount * today_unit_price:  
            units += buy_grid_amount  
            remain_cash -= buy_grid_amount * today_unit_price  
            direction.iloc[i] = 1  
            size.iloc[i] = buy_grid_amount  
  
            # 重置调整基准  
            base_unit_price = today_unit_price  
            position_grid += 1  
  
    return direction, size, grid_interval

选取和之前相同的参数模拟

direction, input_size, grid_interval = stock_grid_line_v2(close,10000,400)

输出：
每格交易价格间隔: 0.6667，每格交易数量: 111
初始买入价格8.7
网格增加数量5

可视化

entries = (direction == 1).vbt.signals.fshift() 
exits = (direction == -1).vbt.signals.fshift()
# 传入的每次交易数量也需要移位  
input_size = input_size.shift(1, fill_value=0)
plot_entry_exit(close,entries,exits,basic_line=close.iloc[0],grid_interval=grid_interval)
image.png

回测

def signal_func_nb_v2(c, long_signals, short_signals, size,input_sizes):  
    long_signal = vbt.pf_nb.select_nb(c, long_signals)  
    short_signal = vbt.pf_nb.select_nb(c, short_signals)  
    input_size = vbt.pf_nb.select_nb(c, input_sizes)  
    if long_signal:  
        print(f'buy {input_size} amount')  
        size[0, c.col] = input_size  
    if short_signal:  
        print(f'sell {input_size} amount')  
        size[0, c.col] = input_size  
    return long_signal, False, short_signal, False  
pf = vbt.Portfolio.from_signals(  
    close=close,   
    init_cash=10000, # 初始资金  
    init_position= 400, #初始仓位  
    init_price = close.iloc[0], #初始建仓金额  
    signal_func_nb=signal_func_nb_v2,  
    signal_args=(  
        vbt.Rep("long_signals"),   
        vbt.Rep("short_signals"),   
        vbt.Rep("size"),  
        vbt.Rep("input_sizes")      
    ),  
    size=vbt.RepEval("np.full((1, wrapper.shape_2d[1]), np.nan)"),    
    size_type="amount",  
    accumulate=True,  
    broadcast_named_args=dict(  
        long_signals= entries,  
        short_signals=exits,  
        input_sizes=input_size,  
    ),  
    jitted = False  
)
pf.stats()

image.png
就回报率而言没有太大变化，但是其他指标有所提升。

值得注意的是，这个策略的超参数很多，网格上下限，网格数量，准备投入金额，初始建仓数量，都可以选取改变。况且就股票市场来说，买入的数量基本都是以1手为单位，上述的111交易份额实际不符合预期。

策略也有很多优化空间，比如入场时机或许可以选择一些参数联合判断（如 EMA ，RSI 等）。

Vectorbt 的优势就在于能够快速的遍历测试大规模数据的情况，进而从中选取合适的超参数。这里限于篇幅和进度，先暂时分享上述内容。