Python搭建轻量级量化策略系统

1.策略容器
# -*- coding: utf-8 -*-
from userStrategy import userconfig
config = {
"mod": {
"stock": {
"enabled": True,
},
"future": {
"enabled": False,
}
}
}
from events import EventBus
class Env(object):
_env = None
def __init__(self, config):
Env._env = self
self.config = config
self.event_bus = EventBus()
self.usercfg = userconfig
self.global_vars = None
self.current_dt = None

@classmethod
def get_instance(cls):
"""
返回已经创建的 Environment 对象
"""
if Env._env is None:
raise RuntimeError("策略还未初始化")
return Env._env

def set_global_vars(self, global_vars):
self.global_vars = global_vars2.事件及事件注册机
# -*- coding: utf-8 -*-
from enum import Enum
from collections import defaultdict

class Event(object):
def __init__(self, event_type, **kwargs):
self.__dict__ = kwargs
self.event_type = event_type

def __repr__(self):
return ' '.join('{}:{}'.format(k, v) for k, v in self.__dict__.items())

class EventBus(object):
def __init__(self):
self._listeners = defaultdict(list)

def add_listener(self, event, listener):
self._listeners[event].append(listener)

def prepend_listener(self, event, listener):
self._listeners[event].insert(0, listener)

def publish_event(self, event):
for l in self._listeners[event.event_type]:
# 如果返回 True ，那么消息不再传递下去
if l(event):
break
class EVENT(Enum):
# 股票
STOCK = 'stock'
# 期货
FUTURE = 'future'
def parse_event(event_str):
return EVENT.__members__.get(event_str.upper(), None)3.业务处理模块加载及handler
# -*- coding: utf-8 -*-

from collections import OrderedDict
from importlib import import_module

class ModHandler(object):
def __init__(self):
self._env = None
self._mod_list = list()
self._mod_dict = OrderedDict()

def set_env(self, environment):
self._env = environment

config = environment.config

for mod_name in config['mod']:
if config['mod'][mod_name]['enabled'] == False:
continue
self._mod_list.append(mod_name)

def start_up(self):
for mod_name in self._mod_list:
#动态加载模块
mod = import_module(mod_name)
for i in mod.__all__:
i(self._env.event_bus)

# -*- coding: utf-8 -*-
import random
from events import EVENT
__all__ = [
]
def export_as_api(func):
__all__.append(func)
return func

@export_as_api
def startup(event_bus):
event_bus.add_listener(EVENT.STOCK, handler)
print('load and register stock mod')

def handler(event):
rd = random.randint(0,9)
if rd == 7:
print('成交记录：UserTrade({''secu:000001.XHSG,''order_id'': 1522310538, ''trade_id'': 1522310538, ''price'': 10.52, ''amount'': 2300')
if rd == 5:
print('成交记录：UserTrade({''secu:000254.XHSG,''order_id'': 1522310538, ''trade_id'': 1522310538, ''price'': 23.52, ''amount'': 1700')
if rd == 2:
print('成交记录：UserTrade({''secu:600012.XHSG,''order_id'': 1522310538, ''trade_id'': 1522310538, ''price'': 13.52, ''amount'': 1700')
pass

4.处理策略文本主程序
# -*- coding: utf-8 -*-
from Env import Env, config
from Mod import ModHandler
from events import EVENT, Event
from globalVars import GlobalVars
from CodeLoader import CodeLoader
def run_file(strategy_file_path):
#加载config到env
env = Env(config)
#启动加载模块
mod = ModHandler()
#加载模块中注入config
mod.set_env(env)
#启动加载
mod.start_up()

loader = CodeLoader(strategy_file_path)
scope = {}
scope = loader.load(scope)
env.set_global_vars(GlobalVars())
scope.update({
"g": env.global_vars
})
f = scope.get('initialize', None)
f()
5.类加载器
import codecs
import copy
from six import exec_

class CodeLoader:
def __init__(self, strategyfile):
self._strategyfile = strategyfile

def compile_strategy(self, source_code, strategyfile, scope):
code = compile(source_code, strategyfile, 'exec')
exec_(code, scope)
return scope
#
def load(self, scope):
with codecs.open(self._strategyfile, encoding="utf-8") as h:
source_code = h.read()
source_code = 'from api import *\n' + source_code
return self.compile_strategy(source_code, self._strategyfile, scope)6.api及全局g值
import six
import pickle
class GlobalVars(object):
def get_state(self):
dict_data = {}
for key, value in six.iteritems(self.__dict__):
try:
dict_data[key] = pickle.dumps(value)
except Exception as e:
print('CCC')
return pickle.dumps(dict_data)

def set_state(self, state):
dict_data = pickle.loads(state)
for key, value in six.iteritems(dict_data):
try:
self.__dict__[key] = pickle.loads(value)
except Exception as e:
print('CCC')

# -*- coding: utf-8 -*-
from Env import Env
import datetime
from  events import Event, EVENT

def run_daily(func, time):
env = Env.get_instance()
event_bus = env.event_bus
if time == 'open':
start = datetime.datetime.strptime(env.usercfg['start'], "%Y-%m-%d")
end = datetime.datetime.strptime(env.usercfg['end'], "%Y-%m-%d")
for i in range((end - start).days+1):
day = start + datetime.timedelta(days=i,hours=9,minutes=30)
print(day)
event_bus.publish_event(Event(EVENT.STOCK))
func()
pass

7.测试，编写策略脚本
userconfig = {
"start":"2018-01-01",
"end":"2018-03-29"
}

def initialize():
g.today = "2018-03-23"
#m
run_daily(market_open, time='open')

def market_open():
print("market_open")
pass8.创建运行指定位置的用户策略的脚本
# -*- coding: utf-8 -*-
from main import run_file
file_path = "./userStrategy.py"
run_file(file_path)9.9.测试结果：





10.目录清单