使用下面的代码，进行批量遗传算法参数优化的过程中，内存会随着回测品种的增加不断累积，最终导致程序崩溃。测试机器是16核32进程，64G内存。
有什么好的办法，可以使得每完成一个合约的参数优化之后自动释放内存么？
#%%
from vnpy.trader.optimize import OptimizationSetting
from vnpy_ctastrategy.backtesting import BacktestingEngine
from vnpy_ctastrategy.strategies.atr_rsi_strategy import (
AtrRsiStrategy,
)
from vnpy_ctastrategy.strategies.inverse_strategy import InverseStrategy
from vnpy_ctastrategy.strategies.TrendTrading_v1 import TrendTrading_v1
from vnpy_ctastrategy.strategies.TrendTrading_v2 import TrendTrading_v2
from datetime import datetime, date
import json
import pandas as pd
from string import digits
import os
#%%
class BatchCTAOptimize:
def __init__(
self,
vt_symbol,
strategyname,
vt_symbolconfig='vt_symbols.json',
start_date=datetime(2019, 1, 1),
end_date=datetime(2019,4,30)
):
config = open(vt_symbolconfig)
self.setting = json.load(config)
self.start_date = start_date
self.end_date = end_date
self.vt_symbol = vt_symbol
self.strategyname = strategyname
def addParameters(self,engine,interval="1m", capital=100*10000):
vt_symbol=self.vt_symbol
if vt_symbol in self.setting:
engine.set_parameters(
vt_symbol,
interval=interval,
start=self.start_date,
end=self.end_date,
rate=self.setting[vt_symbol]["fee_rate"],
slippage=self.setting[vt_symbol]["slippage"],
size=self.setting[vt_symbol]["contract_size"],
pricetick=self.setting[vt_symbol]["pricetick"],
capital=capital
)
else:
print("symbol %s hasn't be maintained in config file" % vt_symbol)
return engine
def get_strategy_parameters(self):
engine = BacktestingEngine()
engine = self.addParameters(engine)
engine.add_strategy(eval(self.strategyname), {})
print(engine.parameters)
def strategy_ga_optimize(self):
vt_symbol = self.vt_symbol
engine = BacktestingEngine()
engine = self.addParameters(engine)
engine.add_strategy(eval(self.strategyname), {})
engine.load_data()
setting = OptimizationSetting()
setting.set_target("sharpe_ratio")
setting.add_parameter("atr_length", 25, 27, 1)
setting.add_parameter("atr_ma_length", 10, 30, 10)
result = engine.run_ga_optimization(setting,max_workers= 15,output=False)
for i in range(len(result)):
try:
result[vt_symbol][f"parameters_{i}"] = result[i][0]
result[vt_symbol][f"sharp_ratio_{i}"] = result[i][1]
# result[vt_symbol][f'details_{i}'] = result[i][2]
except:
pass
df = pd.DataFrame.from_dict(result)
path = f'.\\{self.strategyname}'
if not os.path.exists(path):
os.makedirs(path)
df.to_csv(path + f'\\{self.vt_symbol}_{self.strategyname}.csv')
return result
#%%
if __name__ == '__main__':
strategy_name = 'AtrRsiStrategy'
start_date=datetime(2019, 1, 1)
end_date=datetime(2022, 3, 1)
print(eval(strategy_name).parameters)
result = {}
varity_totest = pd.read_csv('dom_condetail.csv',encoding='utf-8-sig')
vt_symbolconfig=r'./vt_symbols.json'
config = open(vt_symbolconfig)
settings = json.load(config)
for vt_symbol in settings:
if vt_symbol.split('.')[0].rstrip(digits) not in varity_totest.code.values:
continue
bt = BatchCTAOptimize(vt_symbol,strategyname=strategy_name,start_date=start_date, end_date=end_date)
r = bt.strategy_ga_optimize()
result[vt_symbol] = r