2046 大盘股也能穿越牛熊市.py 精品量化交易策略

# 标题：大盘股也能穿越牛熊市 # 作者：曹经纬

'''
大盘的五种走势：
反弹、持续上涨、回调、持续下跌、震荡
'''

from kuanke.wizard import *
from jqdata import *
import datetime
import time
import enum
import operator
import numpy as np
import math
import random

import talib
import jqdata as jy

from csv import DictReader
from sys import version_info
from six import StringIO

# 仓位管理
g_symbol_count = 50
g_max_cash_rate = 1.0
#g_balance_rate = min(1.0, g_max_cash_rate / max(g_symbol_count, 4) * 3.5)
g_balance_rate = 0.33
g_balance_rate_per_symbol = g_balance_rate * 1
g_min_available_cash_rate = 1.0 - g_max_cash_rate * 0.95
g_random_index = 1

    
g_benchmark_symbol = '000300.XSHG'
g_log_path = 'log/大盘精准预测.log'


# 初始化函数，设定基准等等
def initialize(context):
    set_benchmark('000300.XSHG')
    set_option('use_real_price', True)
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    
    clear_log(g_log_path)
    initialize_quant_env(context)
    run_daily(timer_event, time='14:30')
    
    
def initialize_quant_env(context):
    g.strategy_manager = StrategyManager()
    g.risk_control = RiskControl()
    g.stock_selector = StockSelector()
    
    g.strategy_manager.ResetStrategy(context, select_random_symbol(g.stock_selector.check_stocks(context), g_random_index, g_symbol_count))

def timer_event(context):
    log.info('+----------+----------+----------+----------+----------+----------+----------+')
    #log.info(datetime_to_string(context.current_dt))
    
    if (context.current_dt.minute % 30 != 0):
        return

    g.risk_control.CheckForBenchmark(context, g_benchmark_symbol)
    
    could_trade = g.risk_control.could_trade_baodie or g.risk_control.could_trade_chaodi
    if not could_trade and len(list(context.portfolio.positions.values())) == 0 and g.strategy_manager.status == 'working':
        if context.current_dt.month % 3 == 0:
            g.strategy_manager.ResetStrategy(context, select_random_symbol(g.stock_selector.check_stocks(context), g_random_index, g_symbol_count))
            g.strategy_manager.status == 'waiting'
        
    
    g.strategy_manager.HandleData(context)    
    if could_trade:
        g.strategy_manager.status = 'working'
        
    if could_trade:
        log.info('%s 大盘预测[多]' % datetime_to_string(context.current_dt))
    else:
        log.info('%s 大盘预测[空]' % datetime_to_string(context.current_dt))


# =============================================================================================
class EnumOrderType(enum.Enum):
    BuyLong = 1
    SellLong = 2
    BuyShort = 3
    SellShort = 4
    
class OrderInfo(object):
    def __init__(self):
        self.strategy_id = ''
        self.side = 'long'
        self.datetime = string_to_datetime('1990-01-01 00:00:00')
        self.amount = 0
        self.price = 0.0


class StrategyBase(object):
    def __init__(self, symbol, index):
        self.symbol = symbol
        self.strategy_index = index

        self.release_price = 0.0 # 发行价
        self.max_price = 0.0
        self.min_price = 10000000.0
        self.first_price = 0.0
        self.last_price = 0.0
        self.first_time = string_to_datetime('1990-01-01 00:00:00')
        self.last_time = string_to_datetime('1990-01-01 00:00:00')
        
        self.order = OrderInfo()
        self.yearly_profit = 0.0
        
        self.value = 0.0 # 股票总市值
        
    @staticmethod
    def GetValuation(symbol, context):
        stmt = query(valuation).filter(valuation.code == symbol)
        rslt = get_fundamentals(stmt, datetime_to_datestring(context.current_dt))
        
        value = 0.0
        if rslt is not None:
            value = rslt['market_cap'][0]
        
        return value

    @staticmethod
    def GetLastPrice(symbol, timeframe):
        vperiod = '1m'
        if (timeframe == 'day'):
            vperiod = '1d'

        #df = attribute_history(symbol, 1, vperiod, ['close'], fq='post')
        df = attribute_history(symbol, 1, vperiod, ['close'])
        if (df['close'] is None):
            return -1.0

        vclose = df['close'][-1]
        return vclose


    def HandleData(self, context):
        last_price = self.GetLastPrice(self.symbol, '1m')
        if math.isnan(last_price) or (last_price <= 0.0):
            return

        if (self.first_price == 0.0):
            self.first_price = last_price

        if (self.first_time == string_to_datetime('1990-01-01 00:00:00')):
            self.first_time = context.current_dt
            
        # TODO 查询太耗时，导致回测时间过长    
        #self.value = self.GetValuation(self.symbol, context)

        self.last_price = last_price
        self.last_time = context.current_dt


        self.max_price = max(self.max_price, self.last_price)
        self.min_price = min(self.min_price, self.last_price)
        if (self.release_price == 0.0):
            self.release_price = self.last_price

        profit = self.last_price / self.first_price
        years = (datetime_to_timestamp(self.last_time) - datetime_to_timestamp(self.first_time)) * 1.0 / (3600 * 24 * 365)
        if (years > 0.1 and profit > 0.0):
            self.yearly_profit = math.pow(profit, 1.0 / years) - 1.0
        
        
    def OpenOrder(self, context, balance_rate, order_side):
        if self.symbol in [position.security for position in list(context.portfolio.positions.values())]:
            return
        
        vbalance = context.portfolio.total_value
        available_cash = context.portfolio.available_cash
        if available_cash < vbalance * balance_rate:
            return
        
        amount = 0
        vorder = order_target_value(self.symbol, vbalance * balance_rate, side=order_side)
        if (vorder is not None and vorder.status == OrderStatus.done):
            write_log(g_log_path, '%s buy  order[%s %s %d] ok.' % (datetime_to_string(context.current_dt), self.symbol, order_side, amount))
            self.order.amount = amount
            self.order.datetime = context.current_dt
            self.order.price = self.last_price
        else:
            write_log(g_log_path, '%s buy  order[%s %s %d] error.' % (datetime_to_string(context.current_dt), self.symbol, order_side, amount))
            

    def CloseOrder(self, context):
        if self.symbol not in [position.security for position in list(context.portfolio.positions.values())]:
            return
        
        vorder = order_target_value(self.symbol, 0, side=self.order.side)
        if (vorder is not None and vorder.status == OrderStatus.done):
            write_log(g_log_path, '%s sell order[%s %s %d] ok.' % (datetime_to_string(context.current_dt), self.symbol, self.order.side, self.order.amount))
        else:
            write_log(g_log_path, '%s sell  order[%s %s %d] error.' % (datetime_to_string(context.current_dt), self.symbol, self.order.side, self.order.amount))      


class StrategyTurtle(StrategyBase):
    def __init__(self, symbol, index):
        super().__init__(symbol, index)
        self.security_info = get_security_info(symbol)
        
        
    def HandleData(self, context):
        super().HandleData(context)
        if (self.first_price <= 0.0):
            return
        
        
        could_trade = g.risk_control.could_trade_baodie or g.risk_control.could_trade_chaodi
        
        last_price = self.GetLastPrice(self.symbol, '1d')
        diff_years = (datetime.date(context.current_dt.year, context.current_dt.month, context.current_dt.day) - self.security_info.start_date).days / 365
        
        #could_trade_price = diff_years > 3 and last_price > 30 * pow(1.2, diff_years - 3)
        could_trade_price = last_price > 50 
        
        could_trade = could_trade and could_trade_price
        
        
        if (not(could_trade)):
            self.CloseOrder(context)
            return
        
        if could_trade:
            if self.CheckForChannel(context, '1d', 60, EnumOrderType.SellLong):
                self.CloseOrder(context)
                
            if self.CheckForChannel(context, '1d', 60, EnumOrderType.BuyLong):
                self.OpenOrder(context, g_balance_rate, 'long')
            elif g.risk_control.could_trade_chaodi:
                self.OpenOrder(context, g_balance_rate, 'long')
        #else:
        #    if self.CheckForChannel(context, '1d', 60, EnumOrderType.SellLong):
        #        self.CloseOrder(context)
                

    def CheckForChannel(self, context, timeframe, period, order_type):
        df = attribute_history(self.symbol, period + 1, timeframe, ['high', 'low', 'close'])
        if (df.shape[0] < period):
            return False

        close_list = df['close']
        if (math.isnan(close_list[0]) or math.isnan(close_list[-1])):
            return False
            
        vhigh = talib.MAX(df['high'], timeperiod=period)[-1]
        vlow = talib.MIN(df['low'], timeperiod=int(period/2))[-1]
        atr = talib.ATR(df['high'], df['low'], df['close'], timeperiod=period)[-1]
        
        channel_upper = vhigh - 2.00 * atr
        channel_lower = vlow  + 0.25 * atr
        
        diff_time = context.current_dt - self.order.datetime
        held_days = diff_time.days
        
        could_trade = False
        if (order_type == EnumOrderType.BuyLong):
            if (self.last_price > channel_upper):
                could_trade = True
                
        elif (order_type == EnumOrderType.SellLong):
            if self.order.price > 0.0:
                profit_rate = (self.last_price - self.order.price) / self.order.price
                loss_rate = (max(self.order.price, vhigh) - self.last_price) / self.order.price
    
                if (loss_rate > 0.15 and held_days > 30) or held_days > 90:       
                    could_trade = True

                
        return could_trade
        
class StrategyManager(object):
    def __init__(self):
        self.strategies = []
        self.status = 'waiting' # waiting working

     
    def ResetStrategy(self, context, strategies):
        self.strategies = []
        for i in range(len(strategies)):
            self.AddStrategy(context, StrategyTurtle(strategies[i], i))

       
    def AddStrategy(self, context, strategy):
        self.strategies.append(strategy)
        write_log(g_log_path, '%s add strategy[%s]' % (strategy.symbol, datetime_to_string(context.current_dt)))

      
    def HandleData(self, context):
        for strategy in self.strategies:
            strategy.HandleData(context)

class IndicatorInfo(object):
    def __init__(self, timeframe, period):
        self.timeframe = timeframe
        self.period = period
        
class RiskControl(object):
    def __init__(self):
        self.could_trade_baodie = False
        self.could_trade_chaodi = False
        
    def CheckForVelocity(self, symbol, period, velocity_min, velocity_max):
        hst = attribute_history(symbol, period, '1d', ['close'])
        close_list = hst['close']
        
        if (len(close_list) == 0):
            return False
        
        if (math.isnan(close_list[0]) or math.isnan(close_list[-1])):
            return False
        
        ma = talib.MA(close_list, timeperiod=period)[-1]
        velocity = (close_list[-1] - close_list[0]) / ma / period
        
        #record(vlc=(velocity-velocity_min)*100)
        return velocity_min <= velocity <= velocity_max
        
    def CheckForBenchmark(self, context, symbol):
        self.could_trade_baodie = self.CheckForVelocity(symbol, 5, -0.008, 0.10)  and check_for_rsi(symbol, '1d', 30, 47.5, 100)