空中花园策略实时信号加工
由bq7zuymm创建，最终由bq7zuymm 被浏览 32 用户
在量化交易中，通过手段实时获取交易信号是基本功，本文将利用dai.stream_factor和其他第三方库配合给你的qq邮件发送策略信号。以下涉及到的流数据表需要权限才可使用，开通权限请咨询小Q。
策略描述
隔夜跳开回一个重要的因子**（我们不做夜盘）**, 若当日高开, 则当日高开高走的可能性较大; 反之若低开, 则当日低开低走可能性较大, 基于这一观点, 我们来开发一个隔夜跳开的策略, 由于跳开形态的存在, 我们将这一类的策略描述为空中花园策略. 该策略分为以下几个步骤:
- 确定交易品种：我们选用的期货品种标的为: "IM2503.CFE"；
- 确定跳开状态：今开价大于昨收价则为上涨跳开, 若今开价小于昨收价则为下跌跳开. 通常我們需要设定两个个阈值，今开价与昨收价差值大于阈值为上涨跳开，反之小于另一个阈值为下跌跳开；
- 确定交易阈值：对于上涨跳开的情况，以当天第二个分钟时刻将分钟最高价对位阈值， 后续价格突破阈值则触发买入多仓的信号。对于下跌跳开的情况， 以当天第二个分钟时刻将分钟最底价对位阈值, 若后续价格突破阈值, 则触发买入空仓信号。
- 下午尾盘清仓：我们下午尾盘预留五分钟平掉我们的仓位。
我们可以看到23号的收盘价低于24号早盘的收盘价，如果我们将阈值定位10的话这是一个显著的上涨跳开，因为3228-3182=46>10。所以我们在开盘的第二根k线上取最高价作为上限，而第三根k线突破了上限加上阈值，所以我们应该开一笔多仓。
我们可以看到20号收盘价显著高于第二日第一个分钟的收盘价，所以这是一个下跌跳开(如果我们选取的阈值为1的话，上图情形下跌了6元)。之后我们在当天第二根分钟k线上取最低点作为下限，之后由于第三根k线收盘价突破了下限减阈值，所以开空仓。
策略代码
导入相关库以及推送数据至中间表
import dai
import time
import smtplib
import numpy as np
import pandas as pd
from datetime import datetime
from email.header import Header
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
dai.pull_data_to_table(datasource='cn_future_bar1m', table_name='future_data', overwrite=True, lookback_time=8*24*60*60)
接下来我们需要获取我们需要的数据:
- 跳开状态: 需要用到收盘价、上一时刻的收盘价;
- 交易阈值: 需要用到最高价、最低价;
- 标的凭证: 我们确定了交易标的, 所以需要证券代码字段；
- 需要再开盘第一分钟确定跳开状态，在第二分钟确定阈值, 所以需要将时间戳转化为分钟数据.
根据以上分析, 我们可以写出一份抽取数据的引擎:
sql = """
SELECT date_trunc('minute', to_timestamp(datetime * 1.0 / 1000 + 8 * 60 * 60)) as date,
instrument, close, high, low, open,
lag(close, 1) OVER (PARTITION BY instrument ORDER BY datetime) AS pre_close, datetime
FROM future_data WHERE instrument = 'IM2503.CFE'
"""
future_data = dai.stream_factor(sql, 'base', True, 'datetime ASC')
引擎冷启动, 以防取到空数据:
for _ in range(1000):
d = future_data.df()
if len(d) != 0:
break
接下来初始化各种参数, 包括当天的跳开状态、上下限等参数
# 由于是实时更新数据，如果时间过了09点31的话我们就无法获取当日的跳开情况了, 所以为了避免这种情况, 我们首先根据历史数据确定跳开情况
# 还要避免由于时间的原因无法获取上下界
now = datetime.now().strftime('%Y-%m-%d')
start_date = now + ' 09:31:00'
his_data = future_data.df()
start_data = his_data[his_data['date']==start_date]
if len(start_data) != 0:
if (start_data['close'] / start_data['pre_close']).values - 1 > threshold:
tiaokai_up = True
print('当日上涨跳开')
elif (start_data['close'] / start_data['pre_close']).values < 1 - threshold:
tiaokai_down = True
print('当日下跌跳开')
line_start_date = now + ' 09:32:00'
line = his_data[his_data['date']==line_start_date]
if len(line) != 0:
upper_line = line['high'].iloc[-1]
down_line = line['low'].iloc[-1]
当然我们还需要将交易信号推送给我们的QQ邮箱上, 我们需要推送的信息有当天仓位、当前累计收益、当前标的价格等，所以将所有信息封装到message类中:
def send_email(sender_email, reciever_email, title, content, key):
"""
:params sender_email: 发送者邮箱
:params reciever_email: 接受者邮箱
:params title: 邮件名
:params content: 邮件内容
:params key: smtp服务秘钥(需要在qq邮箱中获取)
"""
# 创建邮件主体对象
email = MIMEMultipart()
# 设置发件人、收件人和主题
email['From'] = sender_email
email['To'] = reciever_email
email['Subject'] = Header(title, 'utf-8')
email.attach(MIMEText(content, 'plain', 'utf-8'))
# 创建SMTP_SSL对象（连接邮件服务器）
smtp_obj = smtplib.SMTP_SSL('smtp.qq.com', 465)
# 通过用户名和授权码进行登录
smtp_obj.login(sender_email, key)
# 发送邮件（发件人、收件人、邮件内容（字符串））
smtp_obj.sendmail(
sender_email,
[reciever_email],
email.as_string()
)
smtp_obj.quit()
class message:
def __init__(self, stream_data, reciever_email, key):
"""
:params stream_data: dai的数据流
:params reciever_email: 接受者的qq邮箱
:params key: smtp服务秘钥(需要在qq邮箱中获取)
"""
self.holding = 0 # 1为持有多仓, -1为持有空仓
self.stream_data = stream_data
# 数据流初始化
self.stream_data.df()
# 初始化邮件参数
self.reciever_email = reciever_email
self.key = key
# 当日价格数据以及信号数据
self.data = pd.DataFrame([])
def settlement(self):
"""
当天累计资金结算
"""
price_data = self.stream_data.df()
close_price = float(price_data.iloc[-1:]['close'].values)
open_price = float(price_data.iloc[-1:]['open'].values)
data = pd.DataFrame({'close': [close_price], 'open': [open_price], 'position': [self.holding]}) # 包含现在的收盘价、开盘价、信号
self.data = pd.concat([self.data, data], axis=0)
self.data.drop_duplicates(inplace=True)
# 计算累计收益
self.cum_return = ((self.data['close'] - self.data['open']) * self.data['position']).sum()
def buy_open(self):
self.holding = 1
def sell_open(self):
self.holding = -1
def close(self):
self.holding = 0
def refresh(self):
self.data = pd.DataFrame([])
def pull_message(self, extra_info=None):
"""
推送信息
"""
import requests
price_data = self.stream_data.df()
date = price_data['date'].iloc[-1].strftime('%Y-%m-%d %H:%M:%S')
price = price_data['close'].iloc[-1]
if self.holding == 1:
text = '多仓'
elif self.holding == -1:
text = '空仓'
else:
text = '无仓位'
content = f'{date} 时刻 {text}, 当前累计收益为{self.cum_return} 当前收盘价为{price} '
if extra_info:
content += extra_info
send_email(
sender_email=self.reciever_email,
reciever_email=self.reciever_email,
title='交易情况',
content=content,
key=self.key
)
接下来我们需要编写一个循环类实现信息的推送:
m = message(future_data, '这里填写你的QQ邮箱', '这里填写QQ邮箱的授权码(如何申请授权码，我会写在附录中)') # 初始化
# 先将当日开盘调出来看看是何种方式跳开
while True:
# 最新数据推送
data = future_data.df()
now = data['date'].iloc[-1].strftime('%H:%M:%S')
last_price = data['close'].iloc[-1]
if now == '09:31:00':
# 早盘刷新交易次数
trade_num = 0
# 开盘确定当天跳开状态
pre_close = data['pre_close'].iloc[-1]
if last_price / pre_close > 1 + threshold:
tiaokai_up = True
elif last_price / pre_close < 1 - threshold:
tiaokai_down = True
elif now == '09:32:00':
# 确定上下级
upper_line = data['high'].iloc[-1]
down_line = data['low'].iloc[-1]
if now >= '09:32:00' and now < '14:50:00':
# 交易时间段
if tiaokai_up and last_price > upper_line and trade_num==0:
# 开多
m.buy_open()
trade_num += 1
elif tiaokai_down and last_price < down_line and trade_num==0:
# 开空
m.sell_open()
trade_num += 1
elif now == '14:50:00':
# 尾盘平仓
m.close()
if tiaokai_up:
extra_info = f'当日上涨跳开, 上限为{upper_line}'
elif tiaokai_down:
extra_info = f'当日下跌跳开, 下限为{down_line}'
else:
extra_info = '价格较上一个交易日无明显波动'
m.settlement() # 当前分钟结算
m.pull_message(extra_info=extra_info) # 信息推送
time.sleep(60) # 60秒休眠时间
实现效果
附录一
我们来看看如何从QQ邮箱中申请授权码的。
首先登录你的邮箱，点开左上角的设置：
接下来点击账户, 进一步找到"POP3/IMAP/SMTP/Exchange/cardDAV/CalDAV服务"窗口项, 这里我已经开过这个服务项了, 读者需自行开启这个服务, 之后即可获取秘钥填入上面的message类的key参数即可。
由于我之前已申请过授权码，若您是初次使用授权码服务，则显示的合上述图片会不一样。但是您选择服务开启即可。