7、策略编写

一、策略说明

1.1、策略位置

• 所有策略位于{project_path}/strategy目录下
• 本来是想让策略适配各种语法，后来发现我想多了
• 目前语法上跟聚宽有些相似

1.2、执行过程

• 策略的执行过程是配合trader模块的，目前本项目只支持中低频策略。
• trader模块初始化上下文后，先执行策略的initialize(context)函数
• 然后根据run_daily或run_interval函数设定的执行时间，定时执行

1.3、代码结构

最简单的策略BuyBuyBuyStrategy.py代码如下，本策略在初始化的时候设定了每天09:30执行buy方法，buy方法每天买一手完美世界(002624.SZ)，使用了全局变量g。

def initialize(context):
    g.wanmei='002624.SZ'
    run_daily(buy, time="09:30")

def buy(context):
    order_buy(g.wanmei, 100)

二、策略语法

2.1、环境变量

2.1.1、数据结构DictObj

DictObj是本项目自定义的一种数据结构，用于把字典换为一个对象，然后既可以用访问字典的方式，也可以用访问对象的方式，来访问其内部元素。

举个例子：

from finhack.core.classes.dictobj import DictObj

me={
    "name":"无所不能的魂大人",
    "english_name":"woldy",
 }

 obj_me=DictObj(me)

 print(obj_me['name'])
 print(obj_me.name)
 obj_me['wechat']='woldywei'
 obj_me.wechat='woldywei'

上面的代码都能执行，也就是说，既可以使用中括号来访问内部元素，也可以使用点来访问。倒也没啥别的，就是这样比较方便

2.1.2、全局变量g

• g是一个DictObj类型的变量，默认是一个空字典，供策略开发者自由使用
• 自动注入到当前运行的策略中，策略中运行的任何位置都可以使用它
• 具体不需要多作解释，需要的时候你自然会用到它

2.1.3、上下文变量context

• context也是一个DictObj类型的变量，主要用于策略的上下文管理，每个策略中的函数都需要传入context

• 与g变量不同，context有一些列具有意义的属性

    context= DictObj({
        'id':'', #本次运行的唯一id
        'universe':[],    #universe，暂时没用到
        'previous_date':None, #universe，上个交易日
        'current_dt':None, #universe，当前交易日
        'args':None, #执行策略时传入的参数
        'trade':DictObj({
            'market':'',  #交易市场
            'model_id':'',  #模型id
            'start_time':'',
            'end_time':'',
            'benchmark':'000001',
            'log_type':'',  #暂时没用到
            'record_type':'',  #暂时没用到
            'strategy':'',  #策略名称
            'order_volume_ratio':1,   #最大成交比例
            'slip':0,  #滑点
            'sliptype':'pricerelated',  #滑点类型
            'rule_list':''  #规则列表，(如涨跌停限制、最大笔数等)
        }),
    
        'account':DictObj({
            'username':'',
            'password':'',
            'account_id':'',
            'open_tax':0,  #买入税费
            'close_tax':0.001,  #卖出税费
            'open_commission':0.0003,  #买入手续费
            'close_commission':0.0003,  #卖出手续费
            'close_today_commission':0,   #当日卖出手续费
            'min_commission':5  #最低手续费
            
        }),
        
        'portfolio':DictObj({
            'inout_cash':0,
            'cash':0,  #现金
            'transferable_cash':0,  #可交易现金，暂时没用到
            'locked_cash':0,
            'margin':0,
            'total_value':0,  #总市值，和账户市值有些重复，目前以这个为主
            'previous_value':0,
            'returns':0,
            'starting_cash':0,
            'positions_value':0,  #持仓市值
            'portfolio_value':0,  #账户市值
            'locked_cash_by_purchase':0,
            'locked_cash_by_redeem':0,
            'locked_amound_by_redeen':0,
            'positions':{
                
            }
        }),
        
        'data':DictObj({ #行情数据源等设置
            'calendar':[],
            'event_list':[],
            'data_source':'file',
            'daily_info':None,
            'dividend':{},
            'quote':None,
            'client':None
        }),
        
        'logs':DictObj({  #这里主要是记录一些日志
            'trade_list':[],
            'order_list':[],
            'position_list':[],
            'return_list':[],
            'trade_returns':[],
            'history':{}
        }),
        'performance':DictObj({ #这个是策略的表现
            'returns':[],
            'bench_returns':[],
            'turnover':[],
            'win':0,
            'win_ratio':0,
            'trade_num':0,
            'indicators':{}
        })
  
    })
  

1.2.3、其他对象

class OrderCost:
    def __init__(self,open_tax=0, close_tax=0.001,open_commission=0.0003, close_commission=0.0003,close_today_commission=0, min_commission=5):
        self.open_tax=open_tax
        self.close_tax=close_tax
        self.open_commission=open_commission
        self.close_commission=close_commission
        self.close_today_commission=close_today_commission
        self.min_commission=min_commission
        pass



class Order():
    def __init__(self,code='',amount='',is_buy=True,side='long',action='',context=None):
        self.code=code
        self.amount=amount
        self.enable_amount=amount
        self.filled=0
        self.info=Data.get_daily_info(code=code,context=context)
        self.price=Data.get_price(code=code,context=context)
        self.order_id = self.generate_order_id()
        self.is_buy=is_buy
        self.cost=0
        self.slip_value=0 #滑点滑掉了多少
        self.last_sale_price=None
        #正常
        self.status=-1 if self.price==None else 1

        

    def generate_order_id(self):
        timestamp = str(int(time.time() * 1000000))
        data = f"{self.code}_{self.amount}_{timestamp}".encode('utf-8')
        hash_object = hashlib.sha256(data)
        order_id = hash_object.hexdigest()
        return order_id
        

class Position():
    def __init__(self,code,amount,enable_amount,last_sale_price):
        self.code=code
        self.amount=amount
        self.enable_amount=enable_amount
        self.last_sale_price=last_sale_price
        self.cost_basis=last_sale_price
        self.total_value=amount*last_sale_price
        self.total_cost=amount*last_sale_price

2.2、策略函数

策略的函数，是在trader对应vendor目录下function.py定义的，目前框架自带的函数有：

init_context(args)：初始化交易上下文环境

• args: 一个包含配置参数的对象

set_benchmark(code)：设置基准代码

• code: 用作基准的股票代码

set_option(key, value)：设置交易选项

• key: 选项的键
• value: 选项的值

set_order_cost(cost, type=None)：设置订单成本

• cost: 成本对象
• type: 订单类型，可选参数

set_slippage(obj, type=None)：设置滑点

• obj: 滑点对象
• type: 订单类型，可选参数

insert_sorted_list(sorted_list, new_element)：将新元素插入排序列表

• sorted_list: 已排序的列表
• new_element: 要插入的新元素

run_interval(func, time, interval='daily', date_list=[])：按指定时间间隔运行函数

• func: 要运行的函数
• time: 执行时间
• interval: 时间间隔，默认为'daily'
• date_list: 日期列表

run_daily(func, time, reference_security=None)：每天运行函数。

• func: 要运行的函数
• time: 执行时间
• reference_security: 参考证券，可选参数

inout_cash(cash, pindex=None)：资金流入流出

• cash: 现金金额
• pindex: 组合索引，可选参数

order_target(security, amount, style=None, side='long', pindex=0, close_today=False)：目标股数下单

• security: 股票代码
• amount: 目标持股数量
• style: 下单风格，可选参数
• side: 买卖方向，默认为'long'
• pindex: 组合索引，默认为0
• close_today: 是否平今，可选参数

compute_cost(value, action='open')：计算成本

• value: 价值
• action: 动作，默认为'open'

order(security, amount, style=None, side='long', pindex=0, close_today=False)：按股数下单

• security: 股票代码
• amount: 股数
• style: 下单风格，可选参数
• side: 买卖方向，默认为'long'
• pindex: 组合索引，默认为0
• close_today: 是否平今，可选参数

order_value(security, value, style=None, side='long', pindex=0, close_today=False)：按价值下单

• security: 股票代码
• value: 价值
• style: 下单风格，可选参数
• side: 买卖方向，默认为'long'
• pindex: 组合索引，默认为0
• close_today: 是否平今，可选参数

order_target_value(security, value, style=None, side='long', pindex=0, close_today=False)：目标价值下单

• security: 股票代码
• value: 目标价值
• style: 下单风格，可选参数
• side: 买卖方向，默认为'long'
• pindex: 组合索引，默认为0
• close_today: 是否平今，可选参数

cancel_order(order)：撤销订单

• order: 订单对象

get_open_orders()：获取未完成订单

get_orders(order_id=None, security=None, status=None)：获取订单信息

• order_id: 订单ID，可选参数
• security: 股票代码，可选参数
• status: 订单状态，可选参数
• get_trades()：获取成交信息

order_buy(security, amount)：买入指定数量的股票

• security: 股票代码
• amount: 股数

order_sell(security, amount)：卖出指定数量的股票

• security: 股票代码
• amount: 股数

log(message, level='info')：记录日志

• message: 日志信息
• level: 日志级别，默认为'info'

load_preds_data(model_id, cache=False)：加载预测数据

• model_id: 模型ID
• cache: 是否使用缓存，默认为False

delete_preds_data(model_id)：删除预测数据

• model_id: 模型ID

sync(context)：同步上下文(实盘使用)

• context: 上下文对象

bind_action(strategy)：绑定策略动作

• strategy: 策略对象

三、策略示例

3.1、小市值策略

'''
finhack trader run --strategy=SmallCapStrategy
'''
import datetime
import os
import random
from finhack.factor.default.factorManager import factorManager
from finhack.market.astock.astock import AStock
## 初始化函数，设定要操作的股票、基准等等
def initialize(context):
    # 设定沪深300作为基准
    set_benchmark('000001.SH')
    # True为开启动态复权模式，使用真实价格交易
    set_option('use_real_price', True) 
    # 设定成交量比例
    set_option('order_volume_ratio', 1)
    # # 股票类交易手续费是：买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5块钱
    set_order_cost(OrderCost(open_tax=0, close_tax=0.001, \
                             open_commission=0.0003, close_commission=0.0003,\
                             close_today_commission=0, min_commission=5), type='stock')
    
    # 为股票设定滑点为百分比滑点                       
    set_slippage(PriceRelatedSlippage(0.00246),type='stock')
    # 持仓数量
    g.stocknum = 5
    # 交易日计时器
    g.days = 0 
    # 调仓频率
    g.refresh_rate = 15
    # 运行函数
    #inout_cash(100000)

    run_daily(trade, time="09:30")
    # run_daily(trade, time="8:05")
    log.info('get code list')
    g.stock_list=AStock.getStockCodeList(strict=False)
    
    g.factors=factorManager.getFactors(['pe_0','MACD_0','peTtm_0','pb_0','totalMv_0'])
    g.factors=g.factors.reset_index()

## 交易函数
def trade(context):
    if g.days%g.refresh_rate == 0:
        #print(context.portfolio.cash)
        sell_list = list(context.portfolio.positions.keys()) 
        if len(sell_list) > 0 :
            for stock in sell_list:
                order_target_value(stock, 0)  
    
        if len(context.portfolio.positions) < g.stocknum :
            Num = g.stocknum - len(context.portfolio.positions)
            Cash = context.portfolio.cash/Num
        else: 
            Cash = 0
    
        ## 选股
        now_date=context.current_dt.strftime('%Y%m%d')
        df = g.factors[g.factors.trade_date==now_date]
        df = g.factors.query(f"pe_0 > 0 & pb_0 < 2 & MACD_0<0 & peTtm_0<pe_0 & pe_0<50 & pe_0>10")
        df=df.sort_values(by='totalMv_0',ascending=True, inplace=False) 

        stock_list = df.head(g.stocknum)['ts_code'].tolist()
        ## 买入股票
        for stock in stock_list:
            if len(context.portfolio.positions.keys()) < g.stocknum:
                order_value(stock, Cash)

        # 天计数加一
        g.days = 1
    else:
        g.days += 1

3.2、AI策略

'''
finhack trader run --strategy=AITopNStrategy --args='{"model_id":"45813be38c1e215dbed056ccc32e38da"}'
'''
import datetime
import os
import random
import json
from finhack.factor.default.factorManager import factorManager
from finhack.market.astock.astock import AStock
from finhack.trainer.trainer import Trainer
from finhack.trainer.lightgbm.lightgbm_trainer import LightgbmTrainer

## 初始化函数，设定要操作的股票、基准等等
def initialize(context):
    # 设定沪深300作为基准
    set_benchmark('000001.SH')
    # True为开启动态复权模式，使用真实价格交易
    set_option('use_real_price', True) 
    # 设定成交量比例
    set_option('order_volume_ratio', 1)
    # # 股票类交易手续费是：买入时佣金万分之三，卖出时佣金万分之三加千分之一印花税, 每笔交易佣金最低扣5块钱
    set_order_cost(OrderCost(open_tax=0, close_tax=0.001, \
                             open_commission=0.0003, close_commission=0.0003,\
                             close_today_commission=0, min_commission=5), type='stock')
    
    # 为股票设定滑点为百分比滑点                       
    set_slippage(PriceRelatedSlippage(0.00246),type='stock')
    # 持仓数量

    
    g.stocknum = int(context.get('args', {}).get('stocknum', 10))
    # 交易日计时器
    g.days = 0 
    # 调仓频率
    g.refresh_rate = int(context.get('args', {}).get('refresh_rate', 10))
    
    run_daily(trade, time="09:30")
    model_id=context.trade.model_id

    preds_data=load_preds_data(model_id)
    clsLgbTrainer=LightgbmTrainer()
    preds=clsLgbTrainer.pred(preds_data,md5=model_id,save=False)
    g.preds=preds


def trade(context):
    if g.days % g.refresh_rate == 0:
        # 获取当前持有的股票列表
        current_holdings = list(context.portfolio.positions.keys())
        # 预测数据中的今日日期
        now_date = context.current_dt.strftime('%Y%m%d')
        # 获取今日的预测数据
        today_preds = g.preds[g.preds['trade_date'] == now_date]

        # 卖出策略：卖出预测净值下降的股票
        for stock in current_holdings:
            filtered_preds = today_preds[today_preds['ts_code'] == stock]['pred']
            if not filtered_preds.empty and filtered_preds.iloc[0] < 1:
            #if today_preds[today_preds['ts_code'] == stock]['pred'].iloc[0] < 1:
                order_sell(stock, context.portfolio.positions[stock].amount)

        # 买入策略：选择预测增长最高的股票
        # 首先，我们过滤出预测增长的股票
        potential_buys = today_preds[today_preds['pred'] > 1]
        # 按预测值排序，选择增长预测最高的股票
        potential_buys = potential_buys.sort_values(by='pred', ascending=False)
        # 确定买入的股票数量
        num_stocks_to_buy = min(g.stocknum - len(current_holdings), len(potential_buys))
        
        # 如果有股票需要买入
        n=0
        if num_stocks_to_buy > 0:
            # 计算每只股票的买入资金
            sync(context)
            cash_per_stock = context.portfolio.cash / num_stocks_to_buy
            # 买入股票
            for i, row in potential_buys.iterrows():
                stock_to_buy = row['ts_code']
                # 如果股票不在当前持仓中，则买入
                if stock_to_buy not in current_holdings:
                    o=order_value(stock_to_buy, cash_per_stock)
                    if o==True:
                        n=n+1
                    if n==num_stocks_to_buy:
                        break

        # 更新交易日计数器
        g.days = 1
    else:
        # 如果不是调仓日，交易日计数器累加
        g.days += 1