基于大宽策略的自动化交易的思路

用户成长系列
标签: #<Tag:0x00007fcf63ce5f60>

(chaoskey) #1

思路简介(很简单, 就三句话)

  1. 将需要的数据, base64编码到仿真日志.

  2. 每天抓取两次最新的仿真日志中的base64编码数据

  3. 根据大宽的数据, 进行自动买卖交易. .

base64编码

由于, 大宽禁用了base64模块, 我特意实现了这个模块

于是只需要将你需要的数据base64编码后打印日志即可. 比如:

print("<predbase64>%s</predbase64>"%Base64.encode("必要的数据"))

下面是Base64的实现


class Base64:
    '''
            Base64 的原理、实现及应用
            
        https://www.jianshu.com/p/dd8bffc91c3d
        
    说明: 由于大宽禁用了Base64的模块,  于是特意实现之.
    '''
    
    codeStr = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"
    
    @staticmethod
    def encode(bs, encoding="utf-8"):
        '''
                对字节序列进行Base64编码
        输入:
            bs   字符串 或 字节序列 (如果如果是字符串则按指定编码转换成字节序列)
        
        输出:
            base64编码后的字符串
        '''
        
        if bs is None or len(bs)==0:
            return None
        
        return Base64.bin2base64str(Base64.str2bin(bs, encoding))

    def decode(string, encoding="utf-8"):
        '''
                对Base64字符串进行解码
        
        输入:
            string   Base64字符串
        
        输出:
            解码后的字节序列或字符串  (默认返回字符串, 如果编码为空则返回字节序列)
        '''
        
        if string is None or len(string)==0:
            return None

        return Base64.bin2str(Base64.base64str2bin(string), encoding)
    
    @staticmethod
    def str2bin(bs, encoding="utf-8"):
        '''
                字节序列转换为二进制字符串
        '''

        if type(bs)==str:
            bs = bs.encode(encoding)
        
        # 将每个字节转换成8位二进制
        def func(b):
            s = bin(b)[2:]
            # 需要补0的长度
            n = 8-len(s)
            if n>0:
                s = "0"*n + s
            return s
        
        return "".join([func(b) for b in bs])
    
    @staticmethod
    def bin2base64str(string):
        '''
                二进制字符串转换为Base64字符串
        '''
        
        # 末尾补齐
        tail = len(string) % 6
        if tail != 0: # 最后剩2bit,需要补4位,剩4位需要补2bit
            tail = 6 - tail
            string += "0"*tail
        
        base64Str = ""
        for i in range(len(string)//6):
            idx = i*6
            s = "0b" + string[idx:idx+6]
            # 二进制转十进制整数
            base64Str += Base64.codeStr[int(s,2)]

        # 需要补=的位数
        n = (len(string)//6)%4
        if n!=0:
            base64Str += "="*(4-n)
        return base64Str
    
    
    @staticmethod
    def base64str2bin(string):
        '''
                Base64字符串转换为二进制字符串
        '''
        
        # 去除末尾的'='
        if "=" in string:
            idx = string.index("=")
            string = string[:idx]
        
        # 将每个字符转换成6位二进制(按base64码)
        def func(c):
            b = Base64.codeStr.index(c)
            s = bin(b)[2:]
            # 需要补0的长度
            n = 6-len(s)
            if n>0:
                s = "0"*n + s
            return s
        
        return "".join([func(c) for c in string])
    
    @staticmethod
    def bin2str(string, encoding="utf-8"):
        '''
                二进制转换为字符串
        '''
        
        cs = []
        for i in range(len(string)//8):
            idx = i*8
            # 按8位切割
            s = "0b" + string[idx:idx+8]
            # 二进制转整数
            cs.append(int(s,2))
        
        bs = bytes(cs)
        if encoding:
            bs = bs.decode(encoding)
        return bs

抓取最新仿真日志中的base64编码数据

此代码无法在大宽中运行, 必须运行在你自己的Python环境中.


import pandas as pd
from six import StringIO
import datetime
import requests
from lxml import html
import base64

def grab_new_prediction(home, notebook_id, login_name, password):
    '''
            从仿真盘抓取预测数据
    参数:
        home 大宽首页 比如: https://bigquant.com
        notebook_id    仿真盘id  
        login_name  你的大宽登陆账号
        password  你的大宽登陆密码
    '''
    
    try:
        session = requests.Session()

        # 登陆大宽
        login_url = "%s/account/login/"%home 
        result = session.get(login_url)
        tree = html.fromstring(result.text)
        authenticity_token = list(set(tree.xpath("//input[@name='csrfmiddlewaretoken']/@value")))[0]
        payload = {
            "login": login_name, 
            "password": password, 
            "csrfmiddlewaretoken": authenticity_token
        }
        result = session.post(
            login_url, 
            data = payload,
            headers = dict(referer=login_url)
        )
        
        log.info("登陆大宽")

        # 打开仿真盘页面
        url = '%s/trade/strategy?notebook_id=%s'%(home, notebook_id)
        result = session.get(
            url, 
            headers = dict(referer = url)
        )
        
        log.info("打开仿真盘页面")

        # 定位最近执行的日志
        tree = html.fromstring(result.text.decode("utf-8").strip())
        elems = tree.findall(".//div[@class='detail-log']/div[@class='bigquant-board']//table[@id='bigquant-strategy-outputs']//a")
        href, time = elems[0].attrib["href"],datetime.datetime.strptime(elems[0].text.replace("n", "").strip(), "%Y-%m-%d %H:%M") 
        
        # 打开目标日志页
        url = home + href
        result = session.get(
            url, 
            headers = dict(referer = url)
        )
        
        log.info("打开最新的策略日志页")

        # 抓取得到已经Base64编码的预测数据(由于代码也在日志中,所以必须用rfind定位base64的位置)
        base64Str = result.text.decode("utf-8").strip()
        i1 = base64Str.rfind("&lt;predbase64&gt;")
        i2 = base64Str.rfind("&lt;/predbase64&gt;")
        if i1<0 or i2<0:
            raise Exception("日志页中没有Base64编码的预测数据")
        base64Str = base64Str[i1+18:i2]
        #print base64Str
        log.info("从日志页中抓取到Base64编码的预测数据")

        # Base64解码 并且获得预测数据
        body = base64.b64decode(base64Str)
        df = pd.read_csv(StringIO(body),index_col="Unnamed: 0")
        
        log.info("Base64解码获得预测数据")
        
        session.close()
        
        return df
    except Exception, e:
        session.close()
        log.warn(str(e))
        return None


自动交易

首先,你必须要有自己的实盘接口或通道, 比如我用的是: 云服务器上部署的基于tradex.dll的接口程序.

一旦获得的大宽策略的必要数据, 就可以根据这个数据通过 实盘接口下单了.

我的下单的基本方式:

开盘前, 9:20下单(没有成交的 9:26撤单); 盘尾: 深市股票14:59竞价卖出, 沪市股票14:59:30市价卖出 .


(大胡子) #2

为你手动点赞!


(zherzhw) #3

请问现在过的实盘借口或通道哪里有呢?券商不是都封了外部借口?


(chaoskey) #4

我用的就是文中提到的实盘通道. 在接受聚宽策略信号进行实盘已经一年多了, 到目前为止运行都很正常.


(zherzhw) #5

有点糊涂了,文中说的是大宽策略(bigquant?),这边说的聚宽,是jointquant么?还是没找到哪里提供实盘通道,能否提供下帮助、介绍者文档或者接口的页面?


(chaoskey) #6

此文只提供思路. 你需要有你自己的实盘通道.

如果你有自己的实盘通道, 可以接受聚宽的信号, 也可以接受大宽的信号.

我的意思是, 通过我自己的实盘通道, 接受聚宽信号进行实盘, 已经有一年时间了.

而我接受大宽信号刚刚开始.

两者不矛盾.

最后强调: 我不提供实盘通道, 需要你自己准备.


(chaoskey) #7

至于我个人用的实盘通道, 我的文中已经给了点线索. 详细我就不说了.