使用遗传算法优化交易策略

如何使用进化论来提高你的交易策略的表现

使用遗传算法优化交易策略

如果您对交易感兴趣,那么您很可能已经看到了一堆基于技术分析指标的策略。遗憾的是,这些标准策略中的大多数都不起作用,需要努力针对某些行业和/或市场条件优化您的策略。

优化策略的一种非常棒的方法是通过使用遗传算法使用自然选择理论。本质上,该算法的工作原理是:

  • 选取 n 个使用某种策略的交易者,并评估他们的表现。
  • 保留最好的 x% 策略(我使用 30%)并消除剩余的 100-x%(听起来很残酷……)。
  • 通过随机生成新交易者,或通过组合(培育)优秀交易者。

听起来有点熟?这是适者生存。我们希望多次运行上述程序以产生最强的交易者,而忽略任何产生平庸结果的人!

例如,这是我通过快速优化简单的移动平均线交叉策略生成的权益曲线。优化是在科技领域的其他公司上进行的,然后我使用微软(MSFT)作为我的测试公司。

使用遗传算法优化交易策略
标准移动平均交叉策略(蓝色)、使用遗传算法优化的移动平均交叉策略(绿色)和买入并持有(橙色)的权益曲线。请注意,优化发生在其他代码上,Microsoft 用于测试。

它显然比基线策略做得更好!通过进一步调整配置,可能会击败买入并持有,但稍后会详细介绍。

在本文中,我将介绍:

  1. 我们要优化的基本策略。
  2. 如何从头开始编写简单的遗传算法。
  3. 遗传算法的结果。

完整的代码将在整篇文章中分成更小的部分以进行解释,您可以在最后找到完整的工作代码🙂

优化的简单策略

出于演示的目的,我们将优化移动平均线交叉策略。很简单,这种策略涉及我们采用价格行为的两条移动平均线(通常是收盘价),并根据这些平均线何时交叉进行交易。

例如,在标准策略中,我们有:

  • 快速移动平均线:收盘价的10 周期简单移动平均线。
  • 缓慢移动平均线:收盘价的20 周期简单移动平均线。

我们在快速移动平均线从慢速移动平均线(开盘价)下方交叉后的第二天进入交易,并在相反的情况下退出。在下图中,根据该策略的规则,绿色阴影区域表示我们在交易中的位置。

使用遗传算法优化交易策略
2022 年适用于 TSLA 的移动平均交叉策略。绿色区域是进行多头交易的区域。

这可以在 Python 中相当简单地编程:

import numpy as np
import pandas as pd
STRATEGY = {
‘fast_ma_type’: ‘simple’,
‘slow_ma_type’: ‘simple’,
‘fast_ma_field’: ‘Close’,
‘slow_ma_field’: ‘Close’,
‘fast_ma_length’: 10,
‘slow_ma_length’: 20,
}
TICKER = ‘MSFT’
NAME = ‘baseline_strat’
LOWER_DATE_FILT = ‘2017-01-01’
def get_ma_cols(df: pd.DataFrame, strat: dict) -> pd.DataFrame:
”’
Add the moving average columns to the dataset, as per the strategy config.
”’
for ma_type in [‘slow’, ‘fast’]:
if strat[ma_type + ‘_ma_type’] == ‘simple’:
df[ma_type] = (
df[strat[ma_type + ‘_ma_field’]]
.rolling(strat[ma_type + ‘_ma_length’])
.mean()
)
elif strat[ma_type + ‘_ma_type’] == ‘exponential’:
df[ma_type] = (
df[strat[ma_type + ‘_ma_field’]]
.ewm(span = strat[ma_type + ‘_ma_length’], adjust = False)
.mean()
)
else:
raise ValueError(
‘There is no current implementation for the ‘ +
strat[ma_type + ‘_ma_type’] + ‘ moving average type.’
)
return df
def run_strat(open_prices: np.array,
fast_ma: np.array,
slow_ma: np.array) -> np.array:
”’
Run the ma crossover strategy. Here, we buy the day after the fast ma
crosses from below the slow ma, and sell when the opposite occurs.
Parameters
———-
open_prices : np.array
The financial instrument open prices on each day
fast_ma : np.array
The faster moving average
slow_ma : np.array
The slower moving average
Returns
——-
trade_res : np.array
The percentage gained/lost on each trade
”’
# Flag to determine whether the instrument is currently held or not
holding = False
# Empty lists to store the results from the strategy
trade_res = []
bought_on = []
sold_on = []
# The logical criteria for if a ma crossover happens, both on the buy and
# sell side
ma_buy = lambda day: (
fast_ma[day2] < slow_ma[day2] and
fast_ma[day1] > slow_ma[day1]
)
ma_sell = lambda day: (
fast_ma[day2] > slow_ma[day2] and
fast_ma[day1] < slow_ma[day1]
)
for day in range(2, open_prices.shape[0]):
if not holding and ma_buy(day):
bought_at = open_prices[day]
bought_on.append(day)
holding = True
elif holding and ma_sell(day):
trade_res.append(open_prices[day]/bought_at 1)
sold_on.append(day)
holding = False
# We are only interested in stats from completed trades, so if we are still
# in a trade we delete the last buy
if holding:
bought_on = bought_on[:1]
return (
np.array(bought_on),
np.array(sold_on),
np.array(trade_res),
)
if __name__ == ‘__main__’:
# Read in the price data and calculate the moving average columns
df = pd.read_csv(f’data/{TICKER}.csv’)
df = get_ma_cols(df, STRATEGY)
# Apply the lower date filter
df = df[df[‘Date’] >= LOWER_DATE_FILT]
# Run the strategy
bought_on, sold_on, trade_res = run_strat(
df[‘Open’].values.astype(np.float64),
df[‘fast’].values.astype(np.float64),
df[‘slow’].values.astype(np.float64),
)
# Form a dataframe with the trading information
dates = df[‘Date’].values
df_backtest = pd.DataFrame({
‘bought_on’: dates[bought_on],
‘sold_on’: dates[sold_on],
‘profit’: trade_res,
})
# Save the price data and backtesting results to a csv for plotting
df_backtest.to_csv(f’{TICKER}_{NAME}_backtest.csv’, index = False)

⚠️注意:您必须在名为“data”的文件夹中包含所选代码的价格数据,该文件夹与上述代码位于同一目录中。⚠️

可以yfinance使用以下代码使用包下载价格数据:

将 yfinance 导入为 yf 
yf.download('MSFT').to_csv('data/MSFT.csv')

那么,这段代码究竟做了什么?

很高兴您问到,此代码将执行以下操作:

  1. 加载价格数据,并生成移动平均线(作为价格数据框中的列)。移动平均线的参数可以在STRATEGY全局变量中找到。
  2. LOWER_DATE_FILT从该日期起仅应用回测。
  3. 运行该策略,并返回购买、出售资产的日期以及每笔交易的利润/损失(以百分比表示)。
  4. 将回测结果保存为主目录中的 csv 文件。

为什么你使用循环而不是矢量化?

这样做有两个原因:

  1. 可读性:循环代码通常更具可读性,尤其是在策略逻辑变得更加复杂的情况下。
  2. 使用 numba:在遗传算法中,您可以使用numba增强循环代码性能的包。

遗传算法

遗传算法可能有点难以理解,最终,它会稍微盯着代码看所有组件是如何组合在一起的。但是,我会将算法分解为关键组件以更详细地描述它们。

阅读后,我的建议是获取完整的代码(在文章末尾),并逐步使用断点来了解每个部分在做什么。如果有什么需要进一步解释,请在下面发表评论,我会解决的🙂

我们可以优化什么?

我已经实现了我们可以优化的策略的六个部分,这些都可以在上面的代码中找到(在STRATEGY全局变量中)。简而言之,这些如下:

  1. 移动平均线类型:简单或指数移动平均线。
  2. 移动平均线字段:基本上,我们在哪个价格字段上取移动平均线(开盘价、低点、高点或收盘价)。
  3. 移动平均长度:计算移动平均的天数。

您可能希望通过添加止损、利润目标、其他指标等来使其更加复杂。可能性实际上是无穷无尽的,但在本指南中,我们将保持简单。

我们究竟在优化什么?

说实话,这实际上取决于您希望您的策略最大化(或最小化)的内容。在本指南中,我们将实现三个不同的优化值:

  1. 平均值:所有交易的平均收益/损失百分比。
  2. 中位数:所有交易的中位数百分比收益/损失。
  3. 复合:您的策略将产生多少倍的初始投资(我个人最喜欢的)。

这就是所谓的适应值。当然您可以选择添加更复杂的适应度值,例如夏普比率、最大损失、最大增益等。

我们将在几个代码上优化策略,因此您最终会得到每个代码的适应度值。这意味着需要一种组合适应度值的方法,因为该算法需要每个策略的单个适应度值。我选择实施:

  1. Min:所有适应度值的最小值。
  2. 平均值:所有适应度值的平均值。
  3. 中位数:所有适应度值的中位数。

我们如何获得新的策略参数?

我已经实现了三种获取新策略参数的方法

  1. 随机选择:随机生成的策略,见get_random_strat函数。
  2. 策略扰动:这会在移动平均长度上加/减几天,并随机交换移动平均类型(请参阅 参考资料perturb_strat)。
  3. 组合制胜策略:对于每个策略参数,我们从前 x% 的策略中获得一份副本(请参阅 参考资料breed_winning_strats)。

笔记!这可能会产生具有不可行参数的策略。换句话说,您最终可能会获得一种策略,其中快速移动平均线的长度比慢速移动平均线长。为了检查这些不一致,check_strat在创建每个新策略时调用该函数。

# Strategy parameters to choose from
MA_TYPES = [‘simple’, ‘exponential’] # Types of moving averages to consider
MA_FIELDS = [‘Open’, ‘Low’, ‘High’, ‘Close’] # Price fields to choose from
LOWER_MA_LENGTH = 3 # The least length a moving average can have
UPPER_MA_LENGTH = 300 # The maximum length a moving average can
MAX_PERTURB = 10 # The maximum number to perturb the strategy parameters with
# The strategy to intially perturb, and also to use as a benchmark during
# the testing phase
STARTING_STRAT = {
‘fast_ma_type’: ‘simple’,
‘slow_ma_type’: ‘simple’,
‘fast_ma_field’: ‘Close’,
‘slow_ma_field’: ‘Close’,
‘fast_ma_length’: 10,
‘slow_ma_length’: 20,
}
def get_random_strat() -> dict:
”’
Generate a fresh random strategy by randomly selecting from the parameters
definied in the global variables.
”’
return check_strat({
‘fast_ma_type’: random.choice(MA_TYPES),
‘slow_ma_type’: random.choice(MA_TYPES),
‘fast_ma_field’: random.choice(MA_FIELDS),
‘slow_ma_field’: random.choice(MA_FIELDS),
‘fast_ma_length’: random.randint(LOWER_MA_LENGTH, UPPER_MA_LENGTH),
‘slow_ma_length’: random.randint(LOWER_MA_LENGTH, UPPER_MA_LENGTH),
})
def check_strat(strat: dict) -> dict:
”’
This checks if the strategy has valid parameters, and adjusts if not. For
example, if the slower moving average has a smaller length than the faster
one, then this is changed to having a larger value.
”’
for ma_type in [‘slow’, ‘fast’]:
if strat[ma_type + ‘_ma_length’] < LOWER_MA_LENGTH:
strat[ma_type + ‘_ma_length’] = LOWER_MA_LENGTH
elif strat[ma_type + ‘_ma_length’] > UPPER_MA_LENGTH:
strat[ma_type + ‘_ma_length’] = UPPER_MA_LENGTH
if strat[‘slow_ma_length’] <= strat[‘fast_ma_length’]:
strat[‘slow_ma_length’] = strat[‘fast_ma_length’] + 1
return strat
def perturb_strat(strat: dict) -> dict:
”’
Perturb the parameters of the strategy slightly to generate a new strategy
”’
for ma_type in [‘slow’, ‘fast’]:
strat[ma_type + ‘_ma_type’] = random.choice(MA_TYPES)
strat[ma_type + ‘_ma_field’] = random.choice(MA_FIELDS)
strat[ma_type + ‘_ma_length’] += (
np.random.randint(MAX_PERTURB, MAX_PERTURB)
)
return check_strat(strat)
def breed_winning_strats(good_strats: np.array,
strats: dict) -> dict:
”’
Taking parameters from good/winning strategies and breed a new strategy.
Parameters
———-
good_strats : np.array
The index values of the best strategies from the evolution
strats : dict
The dictionary of all strategies
”’
new_strat = {}
for param in strats[‘0’].keys():
rand_strat_idx = str(random.choice(good_strats))
new_strat[param] = strats[rand_strat_idx][param]
return check_strat(new_strat)
这是一个代码片段,不会自行运行!完整代码见末尾。

好长的文章!

我知道,所以这是我发现的一张香蕉的随机图片。希望这没有让你感到饥饿。

使用遗传算法优化交易策略

我们如何初始化算法?

我们需要几个组件:

  1. 价格数据:这是一个数据框列表,其中包括我们正在优化的每个代码的价格数据。
  2. 一组策略:这些都是来自基线策略的随机扰动,索引从 0 到NUM_STRATS.
  3. 适应度值:这只是一个二维数组,第一列显示策略索引,第二列显示适应度值(最初设置为零)。
  4. 计算哪些适应度值:为了节省计算时间,我们只计算新生成策略的适应度值(不需要重新计算我们保留的策略)。最初,我们需要计算所有适应度值。
TRAINING_TICKERS = [‘AMZN’, ‘TSLA’, ‘AAPL’, ‘GOOG’, ‘AMD’]
NUM_STRATS = 50 # Number of strategies to try on each evolution
def init_ga() -> Tuple[list, dict, np.array, np.array]:
”’
Initialise the parameters and data needed for the genetic algorithm
Returns
——-
price_data : list
Price data for all tickers we are optimising over
strats : dict
A random set of strategies
fitness : np_arr
An array to store the fitness values in for each strategy
fitness_to_calc : np_arr
An array to indicate which strategies to calculate the fitness for
”’
price_data = [
pd.read_csv(f’data/{ticker}.csv’)
for ticker in TRAINING_TICKERS
]
# Initialise by finding NUM_STRATS strategies which are perturbations from
# the starting strategy defined in the global variables
strats = {
f’{n}: perturb_strat(deepcopy(STARTING_STRAT))
for n in range(0, NUM_STRATS)
}
# Initialise an empty array to store the fitness values in, col 1 is the
# idx value of the strategy, and col 2 stores the fitness value
fitness = np.zeros((NUM_STRATS, 2))
fitness[:, 0] = np.arange(0, NUM_STRATS)
# Initialise the array to determine which strategies to calculate the
# fitness for. Initially its all of them, but in the optimisation we only
# need to calculate for some of them
fitness_to_calc = np.arange(0, NUM_STRATS)
return price_data, strats, fitness, fitness_to_calc
这是一个代码片段,不会自行运行!完整代码见末尾。

那么遗传算法是如何执行的呢?

总而言之,首先我们得到如上所述的初始参数,然后我们确定每次迭代要改变的策略数量。例如,如果我们有 50 个策略,并且每次进化都保留最好的 20%,那么我们需要用新的策略替换 40 个最差的策略。

在此初始化之后,将应用以下过程:

  1. 计算每个新生成策​​略的适应度函数。
  2. 找到前 x% 的策略以保持下一次进化。
  3. 将坏策略分成 3 个(大致)相等的集合以替换为新策略(使用上面解释的三种方法)
  4. 找到需要在下一次进化中进行新适应度计算的策略指标。
  5. 打印出最佳 5 种策略的适应度值。这提供了一种直观的方法来查看优化器是否在工作。
NUM_EVOLVE = 250 # Number of evolutions to perform
KEEP_PERC = 0.3 # Percentage of top models to keep on each evolution
def main() -> dict:
”’
Main running function for the genetic algorithm
Returns
——-
dict
The optimised strategy parameters
”’
# Initialise all the parameters needed to start the evolution
price_data, strats, fitness, fitness_to_calc = init_ga()
# This defines the number of strategies to change on each evolution
num_to_change = int((1KEEP_PERC)*NUM_STRATS)
for evl in range(0, NUM_EVOLVE):
fitness = get_fitness(
price_data,
strats,
fitness,
fitness_to_calc,
)
# Rank the strategies, and select the strategies to change
ranks = fitness[fitness[:, 1].argsort()]
good_strats = ranks[num_to_change:, 0].astype(np.int32)
bad_strats = ranks[:num_to_change, 0].astype(np.int32)
# Split the bad strategies into 3 approx equal sets to make changes
splits = np.array_split(bad_strats, 3)
# Replace some bad strategies with random new ones
for strat in splits[0]:
strats[str(strat)] = get_random_strat()
# Add random perturbations to some good strategies
for strat in splits[1]:
rand_strat = str(random.choice(good_strats))
strats[str(strat)] = perturb_strat(deepcopy(strats[rand_strat]))
# Combine good strategies to make new ones
for strat in splits[2]:
strats[str(strat)] = breed_winning_strats(
good_strats,
deepcopy(strats),
)
# This shows the optimiser which strats have been changed to calculate
# the fitness function on the next iteration. This saves us having to
# recalculate the fitness function for the good strategies and save
# computational time
fitness_to_calc = bad_strats
# Print out evolution statistics for the best five strategies, this
# is helpful to see if the optimiser is doing the correct job (i.e.
# is the fitness being maximised?)
print(f’\nEvolution {evl})
for count, strat in enumerate(np.flipud(good_strats[5:])):
print(
str(count) + ‘. Strategy: ‘ + str(strat) +
‘, ‘ + FITNESS_TYPE + ‘: ‘ +
str(fitness[strat, 1])
)
print(‘———————————————-‘)
# Return the most optimal strategy after all evolutions
return strats[str(good_strats[1])]
这是一个代码片段,不会自行运行!完整代码见末尾。

这不会过拟合吗?

它可以,而且也很容易!使用任何交易策略优化器,您总是会冒这种风险。但是,我们可以引入一些技巧来缓解这种情况!

首先,我们要求算法在每个代码中产生最少 n 笔交易;如果不这样做,算法会严重惩罚该策略。这可以防止它找到一个完美的交易并让它更好地概括。如果优化超过 5 年,那么 20 次交易似乎是一个合理的起点(每年 4 次交易)。

接下来,我们优化了不止一个股票代码,以使其能够泛化到多种工具上。理想情况下,代码越多越好,但这可能会变得非常昂贵,而且您还冒着使数据难以优化的风险。最好选择一个单一的行业,股票以类似的方式移动,然后在那里应用该策略(例如,优化高价科技股,然后不适用于微型股)。

遗传算法结果

为了第一次检查优化器是否正常工作,我决定使用以下配置设置:

  1. NUM_STRATS = 50将 50 名交易员放入锦标赛。
  2. NUM_EVOLVE = 250运行算法的 250 次迭代(演进)。
  3. KEEP_PERC = 0.3在每次迭代中保持最好的 30%(即 15 个交易者)。
  4. STRAT_EVAL = 'compounded'找到每个股票的初始投资的乘数。
  5. FITNESS_TYPE = 'mean'找到所有股票代码的平均乘数。

我对 AMZN、TLSA、AAPL、GOOG 和 AMD 进行了优化,并设置了较低的日期过滤器,2017-01-01以便它们都是高价的大型科技股(粗略地说)。以下是优化器的执行方式及其提供的参数:

使用遗传算法优化交易策略
优化终止后的最终命令窗口输出。

测试是在 MSFT 和 NVDA 上进行的,以显示基线策略和优化策略之间的比较。这有助于确认/否认优化器是否在训练集上过拟合。

显然,这些资产是相关的,可能会产生一些过度拟合。可以肯定的是,需要进行严格而广泛的测试。进一步的步骤是添加一个上限日期过滤器并执行“未来”测试。

适应度值如何随时间演变?

下图显示了每次进化中前 10 种策略的适应度值。显然,所有策略都很快收敛到相同的值(约 20 次演变);在此之后,您会更大幅度地跳到更高的值,这可能是由引入新的随机策略引起的。

使用遗传算法优化交易策略
在对数尺度上绘制的前 10 种策略的适应度值的演变(最大值 = 250 次演变)。

您可以执行超过 250 次进化,并可能获得更好的值。有时几乎会瞬间达到最优值,毕竟这个优化过程是随机的!

我们能做得更好吗?

当然,您只需要稍微修改一下算法参数(我更改了以下内容):

  1. NUM_EVOLVE = 500
  2. FITNESS_TYPE = 'min'找到所有股票代码的最小适应度值。

我的怀疑是,最大化最坏的情况会迫使算法更好地泛化,而且,允许更多的进化为更好的策略出现提供了更高的机会。

该算法推导出以下策略参数:

{'fast_ma_type':'指数','slow_ma_type':'简单','fast_ma_field':'打开','slow_ma_field':'低','fast_ma_length':4,'slow_ma_length':7}

它还生成了这条净值曲线:

使用遗传算法优化交易策略
使用最小优化方法时的权益曲线比较,每 500 次演变中的每一次都有 50 种策略。

这一次,策略实际上跑赢了买入并持有!令我惊奇的是,策略参数显然是人类永远无法想出的,它们看起来不合逻辑,但它们似乎运作良好。

⚠️请记住,以上内容仅用于教育目的并非交易内容的建议。未经自己测试,切勿相信您在互联网上看到的任何东西!⚠️

这种策略可能会过度拟合,并且可能不适用于其他工具。我在此处包含了结果,以表明该算法执行了预期的工作,但最终需要进行严格的测试以确保这是在实际环境中使用的好策略!

感谢您的阅读,希望您喜欢这篇文章!请随时在marslass上与我联系,我很想听听您是否/如何使用代码🙂

完整代码:

下载

 

给TA打赏
共{{data.count}}人
人已打赏
加密交易策略加密学院

止损是什么?如何设置止损点?止损优缺点分析

2022-6-15 16:51:28

加密交易策略

无上交易策略大公开,加减仓技巧最佳技巧-滚动仓位

2022-9-17 14:53:19

0 条回复 A文章作者 M管理员
    暂无讨论,说说你的看法吧
个人中心
购物车
优惠劵
今日签到
有新私信 私信列表
搜索