|
原创文章第543篇，专注“AI量化投资、世界运行的规律、个人成长与财富自由"。 本次改动比较大，底层引擎更换了。 使用更加简洁了，代码量更少了。 1、去除了wxpython壳。 2、使用bt引擎。
3、支持低代码创建与回测策略。 代码已经提交，大家可前往下载，含全市场数据： 
大家下载代码之后，安装好依赖，直接运行main即可： 
写一个策略特别简单： 第一步：选择交易标的，这里咱们选择沪深300和创业板，两个代表大小盘的ETF。
第二步： 新增一个动量因子：
第三步：
交易规则配置：
左边是进场规则 ，右边是离场规则：
第四步：一些回测细节，由于这个策略比较简单，保持默认参数即可：
点击保存策略，或者回测。 
年化20%，还不错哦。
数据在如下位置： 
送一张优惠券： gui代码如下： import os
from dataclasses import asdict
import streamlit as st
st.session_state['rules_buy'] = []
st.session_state['rules_sell'] = []
import requests
import pandas as pd
from config import DATA_DIR
from task import Task, run_task, task_from_json
# from engine.engine import Engine
from datetime import datetime
from . import gui_const
url = 'http:///api/funds'
st.session_state['features'] = []
st.session_state['feature_names'] = []
g_feature_names = st.session_state['feature_names']
g_features = st.session_state['features']
st.session_state['symbols'] = []
g_symbols = st.session_state['symbols']
st.session_state['task'] = Task()
g_rules_buy = st.session_state['rules_buy']
g_rules_sell = st.session_state['rules_sell']
def show_tasks():
directory = DATA_DIR.joinpath('tasks')
# os.listdir() 列出目录下的所有文件和目录名
tasks = []
for filename in os.listdir(directory.resolve()):
file_path = os.path.join(directory, filename)
if os.path.isfile(file_path): # 判断是否是文件
print(f'文件: {file_path}')
tasks.append(filename)
tasks.insert(0, '创建新策略')
task_name = st.selectbox('本地策略列表：', options=tasks, index=0)
if task_name and task_name != '创建新策略':
task_ = task_from_json(filename)
st.session_state['feature_names'] = task_.feature_names
st.session_state['features'] = task_.features
st.session_state['rules_buy'] = task_.rules_buy
st.session_state['rules_sell'] = task_.rules_sell
# st.write(task_)
else:
task_ = Task()
st.session_state['task'] = task_
g_rules_buy = task_.rules_buy
g_rules_sell = task_.rules_sell
# st.write(task_)
@st.cache_data
def get_funds():
datas = requests.get(url).json()
dict_data = {item['symbol']: item['name'] + '({})'.format(item['symbol']) for item in datas}
return dict_data
def select_funds():
def _select_funds_format(x):
return get_funds()[x]
# task.symbols = list(get_funds().keys())
task = st.session_state['task']
symbol_selected = st.multiselect(label='请选择基金：', options=list(get_funds().keys()),
format_func=lambda x: _select_funds_format(x), default=task.symbols)
task.symbols = symbol_selected
def select_period():
task = st.session_state['task']
periods = {'RunDaily': '每天运行', 'RunWeekly': '每周运行', 'RunMonthly': '每月运行', 'RunQuarterly': '每季度运行',
'RunYealy': '每年运行',
'RunDays': '自定义天数'}
def _period_format(x):
return periods[x]
period = st.selectbox(label='请选择调仓周期', index=0, options=list(periods.keys()),
format_func=lambda x: _period_format(x))
if period == 'RunDays':
days = st.number_input(label='请输入天数', min_value=1, max_value=365)
task.algo_period_days = days
task.algo_period = period
def select_weights():
task = st.session_state['task']
weights = {
'WeighEqually': '等权分配',
'WeighFix': '固定权重',
'WeighERC': '风险平价'
}
def _weight_format(x):
return weights[x]
weight = st.selectbox(label='请选择权重分配', options=list(weights.keys()), format_func=lambda x: _weight_format(x))
if weight == 'WeightFix':
fix_weigths = st.text_input(label='请输入权重(逗号分隔)', value='[]')
task.algo_weight_fix = eval(fix_weigths)
task.algo_weight = weight
def add_features():
factors = list(gui_const.ind_dict.keys())
task = st.session_state['task']
g_feature_names = st.session_state['feature_names']
g_features = st.session_state['features']
ind_name = st.selectbox('选择因子', options=factors, key='select_features', index=0)
if st.button('新增'):
ind = gui_const.ind_dict[ind_name]
g_feature_names.append(ind.name)
g_features.append(ind.expr)
count = 0
for i, (name, feature) in enumerate(zip(g_feature_names, g_features)):
col1, col2, col3 = st.columns(3)
with col1:
f = st.text_input(label='因子名', value=name, key=feature + str(count))
g_feature_names[i] = f
with col2:
n = st.text_input(label='表达式', value=feature, key=name + str(count))
g_features[i] = n
with col3:
if st.button('删除', key='btn' + str(count)):
g_feature_names.remove(name)
g_features.remove(feature)
st.rerun()
count += 1
task.features = g_features
task.feature_names = g_feature_names
def add_rule(rules, rule_type='buy'):
task = st.session_state['task']
f = st.selectbox(label='因子', options=task.feature_names, key=rule_type)
if st.button('添加', key='rule_add' + rule_type):
print('f name>>', f)
r = gui_const.Rule(f)
rules.append(r.get_expr())
#st.write(rules)
for i, rule_str in enumerate(rules):
rule = gui_const.Rule().parse_from(rule_str)
col1, col2, col3, col4 = st.columns(4)
with col1:
# st.writee(task.feature_names)
rule.feature_name = st.selectbox(label='因子', options=task.feature_names, key='factor' + rule_type + str(i))
with col2:
op_index = 0
if rule.ops == '=':
op_index = 1
elif rule.ops == "<":
op_index = 2
rule.ops = st.selectbox(label='比较', options=['>', '=', '<'], key='ops' + rule_type + str(i), index=op_index)
with col3:
rule.value = st.number_input(label='值', value=float(rule.value), min_value=-100.0, step=0.01,
max_value=100.0, key='value' + rule_type + str(i))
with col4:
if st.button('删除', key='rule_delete' + rule_type + str(i)):
rules.remove(rule_str)
st.rerun()
rules[i] = rule.get_expr()
def buy_and_sell_rules():
task = st.session_state['task']
g_rules_buy = st.session_state['rules_buy']
g_rules_sell = st.session_state['rules_sell']
col1, col2 = st.columns(2)
with col1:
add_rule(g_rules_buy)
task.rules_buy = g_rules_buy
with col2:
add_rule(g_rules_sell, rule_type='sell')
task.rules_sell = g_rules_sell
def order_by():
task = st.session_state['task']
index = 0
if task.order_by and task.order_by in task.feature_names:
index = task.feature_names.index(task.order_by)
order_by = st.selectbox(label='排序因子', options=task.feature_names, index=index)
top_K = st.number_input(label='top K = ', value=task.topK)
# drop_N = st.number_input(label='排除前N个 = ', value=task.dropN)
task.order_by = order_by
task.topK = top_K
# task.dropN = drop_N
def backtest(task: Task):
if st.button(label='开始回测'):
with st.spinner('回测进行中，请稍后...'):
res = run_task(task)
df = res.prices
ratio = res.stats
orders_df = res.get_transactions()
col1, col2 = st.columns(2)
with col1:
st.write(ratio)
print(ratio)
with col2:
st.line_chart(df)
st.write(orders_df)
def build():
# with col2:
task = st.session_state['task']
tab1, tab2, tab3, tab4 = st.tabs(["选择ETF", "因子配置", "交易规则", "回测设置"])
with tab1:
select_funds()
with tab2:
add_features()
with tab3:
# add_features(task)
buy_and_sell_rules()
with tab4:
col1, col2, col3 = st.columns(3)
with col1:
order_by()
with col2:
select_period()
select_weights()
name = st.text_input(label='策略名称', value=task.name)
task.name = name
col1, col2 = st.columns(2)
with col1:
date_start = st.date_input(label='起始日期', value=datetime(2010, 1, 1))
task.start_date = date_start.strftime("%Y%m%d")
with col2:
date_end = st.date_input(label='结束日期')
task.end_date = date_end.strftime("%Y%m%d")
if st.button(label='保存策略'):
json_data = task.to_json(task.name)
import json
from config import DATA_DIR
with open(DATA_DIR.joinpath('tasks').joinpath(name + '.json'), "w", encoding='UTF-8') as f:
json.dump(asdict(task), f, ensure_ascii=False)
backtest(task)
#st.write(task)
历史文章： 代码发布：AlphaGPT v0.1——基于大模型的智能因子挖掘框架（代码+数据下载）
AI量化实验室——2024量化投资的星辰大海
|
