1630 lines
66 KiB
Python
1630 lines
66 KiB
Python
f"""
|
||
该代码的主要目的是处理Tick数据并生成交易信号。代码中定义了一个tickcome函数，它接收到Tick数据后会进行一系列的处理，包括构建Tick字典、更新上一个Tick的成交量、保存Tick数据、生成K线数据等。其中涉及到的一些函数有：
|
||
on_tick(tick): 处理单个Tick数据，根据Tick数据生成K线数据。
|
||
tickdata(df, symbol): 处理Tick数据，生成K线数据。
|
||
orderflow_df_new(df_tick, df_min, symbol): 处理Tick和K线数据，生成订单流数据。F
|
||
GetOrderFlow_dj(kData): 计算订单流的信号指标。
|
||
除此之外，代码中还定义了一个MyTrader类，继承自TraderApiBase，用于实现交易相关的功能。
|
||
"""
|
||
|
||
# from concurrent.futures import ThreadPoolExecutor
|
||
from multiprocessing import Process, Queue
|
||
import queue
|
||
import threading
|
||
from vnpy.trader.object import TickData, OrderData, TradeData
|
||
from vnpy.trader.constant import Exchange, Direction, Offset, OrderType
|
||
from vnpy.trader.engine import MainEngine
|
||
from vnpy_ctp import CtpGateway
|
||
|
||
# from AlgoPlus.ta.time_bar import tick_to_bar
|
||
import pandas as pd
|
||
from datetime import datetime, timedelta
|
||
from datetime import time as s_time
|
||
import operator
|
||
import time
|
||
import numpy as np
|
||
import os
|
||
import re
|
||
|
||
# import talib as tb
|
||
|
||
import akshare as ak
|
||
|
||
import requests
|
||
|
||
|
||
# # 加入邮件通知
|
||
# import smtplib
|
||
# from email.mime.text import MIMEText # 导入 MIMEText 类发送纯文本邮件
|
||
# from email.mime.multipart import (
|
||
# MIMEMultipart,
|
||
# )
|
||
|
||
# # from email.mime.application import MIMEApplication
|
||
|
||
# # 配置邮件信息
|
||
# receivers = ["240884432@qq.com"] # 设置邮件接收人地址
|
||
# subject = "TD_Simnow_Signal" # 设置邮件主题 订单流策略交易信号
|
||
|
||
# # 配置邮件服务器信息
|
||
# smtp_server = "smtp.qq.com" # 设置发送邮件的 SMTP 服务器地址
|
||
# smtp_port = 465 # 设置发送邮件的 SMTP 服务器端口号，一般为 25 端口 465
|
||
# sender = "240884432@qq.com" # 设置发送邮件的邮箱地址
|
||
# username = "240884432@qq.com" # 设置发送邮件的邮箱用户名
|
||
# password = "osjyjmbqrzxtbjbf" # zrmpcgttataabhjh，设置发送邮件的邮箱密码或授权码
|
||
|
||
tickdatadict = {} # 存储Tick数据的字典
|
||
quotedict = {} # 存储行情数据的字典
|
||
ofdatadict = {} # 存储K线数据的字典
|
||
trade_dfs = {} # pd.DataFrame({}) # 存储交易数据的DataFrame对象
|
||
previous_volume = {} # 上一个Tick的成交量
|
||
tsymbollist = {}
|
||
|
||
clearing_time_dict = {
|
||
"sc": s_time(2, 30),
|
||
"bc": s_time(1, 0),
|
||
"lu": s_time(23, 0),
|
||
"nr": s_time(23, 0),
|
||
"au": s_time(2, 30),
|
||
"ag": s_time(2, 30),
|
||
"ss": s_time(1, 0),
|
||
"sn": s_time(1, 0),
|
||
"ni": s_time(1, 0),
|
||
"pb": s_time(1, 0),
|
||
"zn": s_time(1, 0),
|
||
"al": s_time(1, 0),
|
||
"cu": s_time(1, 0),
|
||
"ru": s_time(23, 0),
|
||
"rb": s_time(23, 0),
|
||
"hc": s_time(23, 0),
|
||
"fu": s_time(23, 0),
|
||
"bu": s_time(23, 0),
|
||
"sp": s_time(23, 0),
|
||
"PF": s_time(23, 0),
|
||
"SR": s_time(23, 0),
|
||
"CF": s_time(23, 0),
|
||
"CY": s_time(23, 0),
|
||
"RM": s_time(23, 0),
|
||
"MA": s_time(23, 0),
|
||
"TA": s_time(23, 0),
|
||
"ZC": s_time(23, 0),
|
||
"FG": s_time(23, 0),
|
||
"OI": s_time(23, 0),
|
||
"SA": s_time(23, 0),
|
||
"p": s_time(23, 0),
|
||
"j": s_time(23, 0),
|
||
"jm": s_time(23, 0),
|
||
"i": s_time(23, 0),
|
||
"l": s_time(23, 0),
|
||
"v": s_time(23, 0),
|
||
"pp": s_time(23, 0),
|
||
"eg": s_time(23, 0),
|
||
"c": s_time(23, 0),
|
||
"cs": s_time(23, 0),
|
||
"y": s_time(23, 0),
|
||
"m": s_time(23, 0),
|
||
"a": s_time(23, 0),
|
||
"b": s_time(23, 0),
|
||
"rr": s_time(23, 0),
|
||
"eb": s_time(23, 0),
|
||
"pg": s_time(23, 0),
|
||
}
|
||
|
||
|
||
# def send_mail(text):
|
||
# msg = MIMEMultipart()
|
||
# msg["From"] = sender
|
||
# msg["To"] = ";".join(receivers)
|
||
# msg["Subject"] = subject
|
||
# msg.attach(MIMEText(text, "plain", "utf-8"))
|
||
# smtp = smtplib.SMTP_SSL(smtp_server, smtp_port)
|
||
# smtp.login(username, password)
|
||
# smtp.sendmail(sender, receivers, msg.as_string())
|
||
# smtp.quit()
|
||
|
||
def send_feishu_message(text):
|
||
headers = {
|
||
"Content-Type": "application/json"
|
||
}
|
||
data = {
|
||
"msg_type": "text",
|
||
"content": {
|
||
"text": text
|
||
}
|
||
}
|
||
response = requests.post("https://open.feishu.cn/open-apis/bot/v2/hook/8608dfa4-e599-462a-8dba-6ac72873dd27", headers=headers, json=data)
|
||
if response.status_code != 200:
|
||
print(f"飞书消息发送失败，状态码: {response.status_code}, 响应内容: {response.text}")
|
||
|
||
def futures_main_day(future_symbol, delta_days):
|
||
# 获取当前日期的数据
|
||
today = datetime.now().strftime("%Y%m%d")
|
||
# 计算多少日前的日期
|
||
start_day = (datetime.now() - timedelta(days=delta_days)).strftime("%Y%m%d")
|
||
|
||
futures_main_sina_hist = ak.futures_main_sina(
|
||
symbol=future_symbol, start_date=start_day, end_date=today
|
||
)
|
||
return futures_main_sina_hist
|
||
|
||
|
||
# 交易程序---------------------------------------------------------------------------------------------------------------------------------------------------------------------
|
||
class ParamObj:
|
||
|
||
symbol = None # 合约名称
|
||
Lots = None # 下单手数
|
||
py = None # 设置委托价格的偏移，更加容易促成成交
|
||
trailing_stop_percent = None # 跟踪出场参数
|
||
fixed_stop_loss_percent = None # 固定出场参数
|
||
dj_X = None # 开仓的堆积参数
|
||
delta = None # 开仓的delta参数
|
||
sum_delta = None # 开仓的delta累积参数
|
||
失衡 = None
|
||
堆积 = None
|
||
周期 = None
|
||
|
||
# 策略需要用到的变量
|
||
cont_df = 0
|
||
pos = 0
|
||
short_trailing_stop_price = 0
|
||
long_trailing_stop_price = 0
|
||
sl_long_price = 0
|
||
sl_shor_price = 0
|
||
out_long = 0
|
||
out_short = 0
|
||
clearing_executed = False
|
||
kgdata = True
|
||
|
||
def __init__(
|
||
self,
|
||
symbol,
|
||
Lots,
|
||
py,
|
||
trailing_stop_percent,
|
||
fixed_stop_loss_percent,
|
||
dj_X,
|
||
delta,
|
||
sum_delta,
|
||
失衡,
|
||
堆积,
|
||
周期,
|
||
):
|
||
self.symbol = symbol
|
||
self.Lots = Lots
|
||
self.py = py
|
||
self.trailing_stop_percent = trailing_stop_percent
|
||
self.fixed_stop_loss_percent = fixed_stop_loss_percent
|
||
self.dj_X = dj_X
|
||
self.delta = delta
|
||
self.sum_delta = sum_delta
|
||
self.失衡 = 失衡
|
||
self.堆积 = 堆积
|
||
self.周期 = 周期
|
||
|
||
|
||
class MyTrader:
|
||
def __init__(self, main_engine, symbol, exchange):
|
||
self.main_engine = main_engine
|
||
self.symbol = symbol
|
||
self.exchange = exchange
|
||
self.param_dict = {}
|
||
self.queue_dict = {}
|
||
self.品种 = " "
|
||
self.tick_subscribed = False
|
||
self.last_tick = None
|
||
self.previous_volume = 0
|
||
|
||
def subscribe_tick(self):
|
||
if not self.tick_subscribed:
|
||
req = self.main_engine.get_gateway("CTP").subscribe(
|
||
[
|
||
{
|
||
"symbol": self.symbol,
|
||
"exchange": self.exchange
|
||
}
|
||
]
|
||
)
|
||
self.tick_subscribed = True
|
||
|
||
def on_tick(self, tick: TickData):
|
||
if self.last_tick is not None:
|
||
last_volume = tick.volume - self.last_tick.volume
|
||
else:
|
||
last_volume = 0
|
||
self.last_tick = tick
|
||
if last_volume == 0:
|
||
return
|
||
tick_dict = {
|
||
"symbol": tick.symbol,
|
||
"created_at": tick.datetime,
|
||
"price": tick.last_price,
|
||
"last_volume": last_volume,
|
||
"bid_p": tick.bid_price_1,
|
||
"bid_v": tick.bid_volume_1,
|
||
"ask_p": tick.ask_price_1,
|
||
"ask_v": tick.ask_volume_1,
|
||
"UpperLimitPrice": tick.upper_limit,
|
||
"LowerLimitPrice": tick.lower_limit,
|
||
"TradingDay": tick.trading_day,
|
||
"cum_volume": tick.volume,
|
||
"cum_amount": tick.turnover,
|
||
"cum_position": tick.open_interest,
|
||
}
|
||
self.process_tick(tick_dict)
|
||
|
||
def process_tick(self, tick):
|
||
# 这里可复用原有的tick处理逻辑
|
||
pass
|
||
|
||
def send_order(self, price, volume, direction, offset):
|
||
order_req = {
|
||
"symbol": self.symbol,
|
||
"exchange": self.exchange,
|
||
"direction": direction,
|
||
"offset": offset,
|
||
"price": price,
|
||
"volume": volume,
|
||
"order_type": OrderType.LIMIT,
|
||
}
|
||
self.main_engine.send_order(order_req, "CTP")
|
||
|
||
# 其他原有方法可继续保留，适配vnpy数据结构
|
||
|
||
def __init__(
|
||
self,
|
||
broker_id,
|
||
td_server,
|
||
investor_id,
|
||
password,
|
||
app_id,
|
||
auth_code,
|
||
md_queue=None,
|
||
page_dir="",
|
||
private_resume_type=2,
|
||
public_resume_type=2,
|
||
):
|
||
self.param_dict = {}
|
||
self.queue_dict = {}
|
||
self.品种 = " "
|
||
|
||
def tickcome(self, md_queue):
|
||
global previous_volume
|
||
data = md_queue
|
||
instrument_id = data["InstrumentID"].decode() # 品种代码
|
||
ActionDay = data["ActionDay"].decode() # 交易日日期
|
||
update_time = data["UpdateTime"].decode() # 更新时间
|
||
|
||
update_millisec = str(data["UpdateMillisec"]) # 更新毫秒数
|
||
created_at = (
|
||
ActionDay[:4]
|
||
+ "-"
|
||
+ ActionDay[4:6]
|
||
+ "-"
|
||
+ ActionDay[6:]
|
||
+ " "
|
||
+ update_time
|
||
+ "."
|
||
+ update_millisec
|
||
) # 创建时间
|
||
# 构建tick字典
|
||
tick = {
|
||
"symbol": instrument_id, # 品种代码和交易所ID
|
||
"created_at": datetime.strptime(created_at, "%Y-%m-%d %H:%M:%S.%f"),
|
||
"price": float(data["LastPrice"]), # 最新价
|
||
"last_volume": (
|
||
int(data["Volume"]) - previous_volume.get(instrument_id, 0)
|
||
if previous_volume.get(instrument_id, 0) != 0
|
||
else 0
|
||
), # 瞬时成交量
|
||
"bid_p": float(data["BidPrice1"]), # 买价
|
||
"bid_v": int(data["BidVolume1"]), # 买量
|
||
"ask_p": float(data["AskPrice1"]), # 卖价
|
||
"ask_v": int(data["AskVolume1"]), # 卖量
|
||
"UpperLimitPrice": float(data["UpperLimitPrice"]), # 涨停价
|
||
"LowerLimitPrice": float(data["LowerLimitPrice"]), # 跌停价
|
||
"TradingDay": data["TradingDay"].decode(), # 交易日日期
|
||
"cum_volume": int(data["Volume"]), # 最新总成交量
|
||
"cum_amount": float(data["Turnover"]), # 最新总成交额
|
||
"cum_position": int(data["OpenInterest"]), # 合约持仓量
|
||
}
|
||
|
||
previous_volume[instrument_id] = int(data["Volume"])
|
||
if tick["last_volume"] > 0:
|
||
self.on_tick(tick)
|
||
|
||
def can_time(self, hour, minute):
|
||
hour = str(hour)
|
||
minute = str(minute)
|
||
if len(minute) == 1:
|
||
minute = "0" + minute
|
||
return int(hour + minute)
|
||
|
||
def on_tick(self, tick):
|
||
# tm = self.can_time(tick["created_at"].hour, tick["created_at"].minute)
|
||
if tick["last_volume"] == 0:
|
||
return
|
||
quotes = tick
|
||
timetick = str(tick["created_at"]).replace("+08:00", "")
|
||
tsymbol = tick["symbol"]
|
||
if tsymbol not in tsymbollist.keys():
|
||
# 获取tick的买卖价和买卖量
|
||
tsymbollist[tsymbol] = tick
|
||
bid_p = quotes["bid_p"]
|
||
ask_p = quotes["ask_p"]
|
||
bid_v = quotes["bid_v"]
|
||
ask_v = quotes["ask_v"]
|
||
else:
|
||
# 获取上一个tick的买卖价和买卖量
|
||
rquotes = tsymbollist[tsymbol]
|
||
bid_p = rquotes["bid_p"]
|
||
ask_p = rquotes["ask_p"]
|
||
bid_v = rquotes["bid_v"]
|
||
ask_v = rquotes["ask_v"]
|
||
tsymbollist[tsymbol] = tick
|
||
tick_dt = pd.DataFrame(
|
||
{
|
||
"datetime": timetick,
|
||
"symbol": tick["symbol"],
|
||
"mainsym": tick["symbol"].rstrip("0123456789").upper(),
|
||
"lastprice": tick["price"],
|
||
"vol": tick["last_volume"],
|
||
"bid_p": bid_p,
|
||
"ask_p": ask_p,
|
||
"bid_v": bid_v,
|
||
"ask_v": ask_v,
|
||
},
|
||
index=[0],
|
||
)
|
||
sym = tick_dt["symbol"][0]
|
||
self.tickdata(tick_dt, sym)
|
||
|
||
def data_of(self, symbol, df):
|
||
global trade_dfs
|
||
trade_dfs[symbol] = pd.concat([trade_dfs[symbol], df], ignore_index=True)
|
||
|
||
def process(self, bidDict, askDict, symbol):
|
||
try:
|
||
# 尝试从quotedict中获取对应品种的报价数据
|
||
dic = quotedict[symbol]
|
||
bidDictResult = dic["bidDictResult"]
|
||
askDictResult = dic["askDictResult"]
|
||
except Exception:
|
||
# 如果获取失败，则初始化bidDictResult和askDictResult为空字典
|
||
bidDictResult, askDictResult = {}, {}
|
||
|
||
# 将所有买盘字典和卖盘字典的key合并，并按升序排序
|
||
sList = sorted(set(list(bidDict.keys()) + list(askDict.keys())))
|
||
|
||
# 遍历所有的key，将相同key的值进行累加
|
||
for s in sList:
|
||
if s in bidDict:
|
||
if s in bidDictResult:
|
||
bidDictResult[s] = int(bidDict[s]) + bidDictResult[s]
|
||
else:
|
||
bidDictResult[s] = int(bidDict[s])
|
||
if s not in askDictResult:
|
||
askDictResult[s] = 0
|
||
else:
|
||
if s in askDictResult:
|
||
askDictResult[s] = int(askDict[s]) + askDictResult[s]
|
||
else:
|
||
askDictResult[s] = int(askDict[s])
|
||
if s not in bidDictResult:
|
||
bidDictResult[s] = 0
|
||
|
||
# 构建包含bidDictResult和askDictResult的字典，并存入quotedict中
|
||
df = {"bidDictResult": bidDictResult, "askDictResult": askDictResult}
|
||
quotedict[symbol] = df
|
||
return bidDictResult, askDictResult
|
||
|
||
def tickdata(self, df, symbol):
|
||
tickdata = pd.DataFrame(
|
||
{
|
||
"datetime": df["datetime"],
|
||
"symbol": df["symbol"],
|
||
"lastprice": df["lastprice"],
|
||
"volume": df["vol"],
|
||
"bid_p": df["bid_p"],
|
||
"bid_v": df["bid_v"],
|
||
"ask_p": df["ask_p"],
|
||
"ask_v": df["ask_v"],
|
||
}
|
||
)
|
||
try:
|
||
if symbol in tickdatadict.keys():
|
||
rdf = tickdatadict[symbol]
|
||
rdftm = pd.to_datetime(rdf["bartime"][0]).strftime("%Y-%m-%d %H:%M:%S")
|
||
now = str(tickdata["datetime"][0])
|
||
if now > rdftm:
|
||
try:
|
||
oo = ofdatadict[symbol]
|
||
self.data_of(symbol, oo)
|
||
if symbol in quotedict.keys():
|
||
quotedict.pop(symbol)
|
||
if symbol in tickdatadict.keys():
|
||
tickdatadict.pop(symbol)
|
||
if symbol in ofdatadict.keys():
|
||
ofdatadict.pop(symbol)
|
||
except IOError as e:
|
||
print("rdftm捕获到异常", e)
|
||
tickdata["bartime"] = pd.to_datetime(tickdata["datetime"])
|
||
tickdata["open"] = tickdata["lastprice"]
|
||
tickdata["high"] = tickdata["lastprice"]
|
||
tickdata["low"] = tickdata["lastprice"]
|
||
tickdata["close"] = tickdata["lastprice"]
|
||
tickdata["starttime"] = tickdata["datetime"]
|
||
else:
|
||
tickdata["bartime"] = rdf["bartime"]
|
||
tickdata["open"] = rdf["open"]
|
||
tickdata["high"] = max(
|
||
tickdata["lastprice"].values, rdf["high"].values
|
||
)
|
||
tickdata["low"] = min(
|
||
tickdata["lastprice"].values, rdf["low"].values
|
||
)
|
||
tickdata["close"] = tickdata["lastprice"]
|
||
tickdata["volume"] = df["vol"] + rdf["volume"].values
|
||
tickdata["starttime"] = rdf["starttime"]
|
||
else:
|
||
print("新bar的第一个tick进入")
|
||
tickdata["bartime"] = pd.to_datetime(tickdata["datetime"])
|
||
tickdata["open"] = tickdata["lastprice"]
|
||
tickdata["high"] = tickdata["lastprice"]
|
||
tickdata["low"] = tickdata["lastprice"]
|
||
tickdata["close"] = tickdata["lastprice"]
|
||
tickdata["starttime"] = tickdata["datetime"]
|
||
except IOError as e:
|
||
print("捕获到异常", e)
|
||
|
||
tickdata["bartime"] = pd.to_datetime(tickdata["bartime"])
|
||
param = self.param_dict[self.品种]
|
||
bardata = (
|
||
tickdata.resample(
|
||
on="bartime", rule=param.周期, label="right", closed="right"
|
||
)
|
||
.agg(
|
||
{
|
||
"starttime": "first",
|
||
"symbol": "last",
|
||
"open": "first",
|
||
"high": "max",
|
||
"low": "min",
|
||
"close": "last",
|
||
"volume": "sum",
|
||
}
|
||
)
|
||
.reset_index(drop=False)
|
||
)
|
||
bardata = bardata.dropna().reset_index(drop=True)
|
||
bardata["bartime"] = pd.to_datetime(bardata["bartime"][0]).strftime(
|
||
"%Y-%m-%d %H:%M:%S"
|
||
)
|
||
tickdatadict[symbol] = bardata
|
||
tickdata["volume"] = df["vol"].values
|
||
self.orderflow_df_new(tickdata, bardata, symbol)
|
||
|
||
def orderflow_df_new(self, df_tick, df_min, symbol):
|
||
# startArray = pd.to_datetime(df_min["starttime"]).values
|
||
voluememin = df_min["volume"].values
|
||
highs = df_min["high"].values
|
||
lows = df_min["low"].values
|
||
opens = df_min["open"].values
|
||
closes = df_min["close"].values
|
||
# endArray = pd.to_datetime(df_min['bartime']).values
|
||
endArray = df_min["bartime"].values
|
||
# print(endArray)
|
||
# deltaArray = np.zeros((len(endArray),))
|
||
# tTickArray = pd.to_datetime(df_tick["datetime"]).values
|
||
bp1minickArray = df_tick["bid_p"].values
|
||
ap1minickArray = df_tick["ask_p"].values
|
||
lastTickArray = df_tick["lastprice"].values
|
||
volumeTickArray = df_tick["volume"].values
|
||
symbolarray = df_tick["symbol"].values
|
||
# indexFinal = 0
|
||
for index, tEnd in enumerate(endArray):
|
||
dt = endArray[index]
|
||
# start = startArray[index]
|
||
bidDict = {}
|
||
askDict = {}
|
||
bar_vol = voluememin[index]
|
||
bar_close = closes[index]
|
||
bar_open = opens[index]
|
||
bar_low = lows[index]
|
||
bar_high = highs[index]
|
||
bar_symbol = symbolarray[index]
|
||
Bp = round(bp1minickArray[0], 4)
|
||
Ap = round(ap1minickArray[0], 4)
|
||
LastPrice = round(lastTickArray[0], 4)
|
||
Volume = volumeTickArray[0]
|
||
if LastPrice >= Ap:
|
||
if str(LastPrice) in askDict.keys():
|
||
askDict[str(LastPrice)] += Volume
|
||
else:
|
||
askDict[str(LastPrice)] = Volume
|
||
if LastPrice <= Bp:
|
||
if str(LastPrice) in bidDict.keys():
|
||
bidDict[str(LastPrice)] += Volume
|
||
else:
|
||
bidDict[str(LastPrice)] = Volume
|
||
# indexFinal = indexTick
|
||
bidDictResult, askDictResult = self.process(bidDict, askDict, symbol)
|
||
bidDictResult = dict(
|
||
sorted(bidDictResult.items(), key=operator.itemgetter(0))
|
||
)
|
||
askDictResult = dict(
|
||
sorted(askDictResult.items(), key=operator.itemgetter(0))
|
||
)
|
||
prinslist = list(bidDictResult.keys())
|
||
asklist = list(askDictResult.values())
|
||
bidlist = list(bidDictResult.values())
|
||
delta = sum(askDictResult.values()) - sum(bidDictResult.values())
|
||
df = pd.DataFrame(
|
||
{
|
||
"price": pd.Series([prinslist]),
|
||
"Ask": pd.Series([asklist]),
|
||
"Bid": pd.Series([bidlist]),
|
||
}
|
||
)
|
||
# df=pd.DataFrame({'price':pd.Series(bidDictResult.keys()),'Ask':pd.Series(askDictResult.values()),'Bid':pd.Series(bidDictResult.values())})
|
||
df["symbol"] = bar_symbol
|
||
df["datetime"] = dt
|
||
df["delta"] = str(delta)
|
||
df["close"] = bar_close
|
||
df["open"] = bar_open
|
||
df["high"] = bar_high
|
||
df["low"] = bar_low
|
||
df["volume"] = bar_vol
|
||
# df['ticktime']=tTickArray[0]
|
||
df["dj"] = self.GetOrderFlow_dj(df)
|
||
ofdatadict[symbol] = df
|
||
|
||
def GetOrderFlow_dj(self, kData):
|
||
param = self.param_dict[self.品种]
|
||
Config = {
|
||
"Value1": param.失衡,
|
||
"Value2": param.堆积,
|
||
"Value4": True,
|
||
}
|
||
aryData = kData
|
||
djcout = 0
|
||
|
||
# 遍历kData中的每一行，计算djcout指标
|
||
for index, row in aryData.iterrows():
|
||
kItem = aryData.iloc[index]
|
||
# high = kItem["high"]
|
||
# low = kItem["low"]
|
||
# close = kItem["close"]
|
||
# open = kItem["open"]
|
||
dtime = kItem["datetime"]
|
||
price_s = kItem["price"]
|
||
Ask_s = kItem["Ask"]
|
||
Bid_s = kItem["Bid"]
|
||
delta = kItem["delta"]
|
||
|
||
price_s = price_s
|
||
Ask_s = Ask_s
|
||
Bid_s = Bid_s
|
||
|
||
gj = 0
|
||
xq = 0
|
||
gxx = 0
|
||
xxx = 0
|
||
|
||
# 遍历price_s中的每一个元素，计算相关指标
|
||
for i in np.arange(0, len(price_s), 1):
|
||
duiji = {
|
||
"price": 0,
|
||
"time": 0,
|
||
"longshort": 0,
|
||
}
|
||
|
||
if i == 0:
|
||
delta = delta
|
||
|
||
order = {
|
||
"Price": price_s[i],
|
||
"Bid": {"Value": Bid_s[i]},
|
||
"Ask": {"Value": Ask_s[i]},
|
||
}
|
||
# 空头堆积
|
||
if i >= 0 and i < len(price_s) - 1:
|
||
if order["Bid"]["Value"] > Ask_s[i + 1] * int(Config["Value1"]):
|
||
gxx += 1
|
||
gj += 1
|
||
if gj >= int(Config["Value2"]) and Config["Value4"] is True:
|
||
duiji["price"] = price_s[i]
|
||
duiji["time"] = dtime
|
||
duiji["longshort"] = -1
|
||
if float(duiji["price"]) > 0:
|
||
djcout += -1
|
||
else:
|
||
gj = 0
|
||
# 多头堆积
|
||
if i >= 1 and i < len(price_s) - 1:
|
||
if order["Ask"]["Value"] > Bid_s[i - 1] * int(Config["Value1"]):
|
||
xq += 1
|
||
xxx += 1
|
||
if xq >= int(Config["Value2"]) and Config["Value4"] is True:
|
||
duiji["price"] = price_s[i]
|
||
duiji["time"] = dtime
|
||
duiji["longshort"] = 1
|
||
if float(duiji["price"]) > 0:
|
||
djcout += 1
|
||
else:
|
||
xq = 0
|
||
|
||
# 返回计算得到的djcout值
|
||
return djcout
|
||
|
||
# 读取保存的数据
|
||
def read_to_csv(self, symbol):
|
||
# 文件夹路径和文件路径
|
||
# 使用正则表达式提取英文字母并重新赋值给symbol
|
||
param = self.param_dict[symbol]
|
||
# symbol = ''.join(re.findall('[a-zA-Z]', str(symbol)))
|
||
folder_path = "traderdata"
|
||
file_path = os.path.join(folder_path, f"{str(symbol)}_traderdata.csv")
|
||
# 如果文件夹不存在则创建
|
||
if not os.path.exists(folder_path):
|
||
os.makedirs(folder_path)
|
||
|
||
# 读取保留的模型数据CSV文件
|
||
if os.path.exists(file_path):
|
||
df = pd.read_csv(file_path)
|
||
if not df.empty and param.kgdata is True:
|
||
# 选择最后一行数据
|
||
# df = df._append(df.iloc[-1], ignore_index=True)
|
||
row = df.iloc[-1]
|
||
|
||
# 根据CSV文件的列名将数据赋值给相应的属性
|
||
param.pos = int(row["pos"])
|
||
param.short_trailing_stop_price = float(
|
||
row["short_trailing_stop_price"]
|
||
)
|
||
param.long_trailing_stop_price = float(row["long_trailing_stop_price"])
|
||
param.sl_long_price = float(row["sl_long_price"])
|
||
param.sl_shor_price = float(row["sl_shor_price"])
|
||
# param.out_long = int(row['out_long'])
|
||
# param.out_short = int(row['out_short'])
|
||
print("找到历史交易数据文件,已经更新持仓，止损止盈数据", df.iloc[-1])
|
||
param.kgdata = False
|
||
else:
|
||
pass
|
||
# print("没有找到历史交易数据文件", file_path)
|
||
# 如果没有找到CSV，则初始化变量
|
||
|
||
pass
|
||
|
||
# 保存数据
|
||
def save_to_csv(self, symbol):
|
||
param = self.param_dict[symbol]
|
||
# 使用正则表达式提取英文字母并重新赋值给symbol
|
||
# symbol = ''.join(re.findall('[a-zA-Z]', str(symbol)))
|
||
# 创建DataFrame
|
||
|
||
data = {
|
||
"datetime": [trade_dfs[symbol]["datetime"].iloc[-1]],
|
||
"pos": [param.pos],
|
||
"short_trailing_stop_price": [param.short_trailing_stop_price],
|
||
"long_trailing_stop_price": [param.long_trailing_stop_price],
|
||
"sl_long_price": [param.sl_long_price],
|
||
"sl_shor_price": [param.sl_shor_price],
|
||
# 'out_long': [param.out_long],
|
||
# 'out_short': [param.out_short]
|
||
}
|
||
|
||
df = pd.DataFrame(data)
|
||
|
||
# 将DataFrame保存到CSV文件
|
||
# df.to_csv(
|
||
# f"traderdata/{str(symbol)}_traderdata.csv",
|
||
# mode="a",
|
||
# index=False,
|
||
# header=False,
|
||
# )
|
||
|
||
traderdata_file_path = f"traderdata/{str(symbol)}_traderdata.csv"
|
||
if os.path.exists(traderdata_file_path):
|
||
# 仅保存最后一行数据
|
||
csv_df = pd.read_csv(traderdata_file_path)
|
||
if df["pos"].iloc[-1] != csv_df["pos"].iloc[-1]:
|
||
df.to_csv(traderdata_file_path, mode="a", header=False, index=False)
|
||
else:
|
||
# 创建新文件并保存整个DataFrame
|
||
df.to_csv(traderdata_file_path, index=False)
|
||
# df.to_csv(f"traderdata/{str(symbol)}_traderdata.csv", index=False)
|
||
|
||
# 每日收盘重置数据
|
||
def day_data_reset(self, symbol):
|
||
param = self.param_dict[symbol]
|
||
sec = "".join(re.findall("[a-zA-Z]", str(symbol)))
|
||
# 获取当前时间
|
||
current_time = datetime.now().time()
|
||
|
||
# 第一时间范围（日盘收盘）
|
||
clearing_time1_start = s_time(15, 5)
|
||
clearing_time1_end = s_time(15, 10)
|
||
|
||
# 创建一个标志变量，用于记录是否已经执行过
|
||
param.clearing_executed = False
|
||
# 检查当前时间第一个操作的时间范围内
|
||
if (
|
||
clearing_time1_start <= current_time <= clearing_time1_end
|
||
and not param.clearing_executed
|
||
):
|
||
param.clearing_executed = True # 设置标志变量为已执行
|
||
trade_dfs[symbol].drop(
|
||
trade_dfs[symbol].index, inplace=True
|
||
) # 清除当天的行情数据
|
||
|
||
# 检查当前时间是否在第二个操作的时间范围内(夜盘收盘)
|
||
elif sec in clearing_time_dict.keys():
|
||
clearing_time2_start = clearing_time_dict[sec]
|
||
clearing_time2_end = s_time(
|
||
clearing_time2_start.hour, clearing_time2_start.minute + 15
|
||
)
|
||
if (
|
||
clearing_time2_start <= current_time <= clearing_time2_end
|
||
and not param.clearing_executed
|
||
):
|
||
param.clearing_executed = True # 设置标志变量为已执行
|
||
trade_dfs[symbol].drop(
|
||
trade_dfs[symbol].index, inplace=True
|
||
) # 清除当天的行情数据
|
||
else:
|
||
param.clearing_executed = False
|
||
pass
|
||
return param.clearing_executed
|
||
|
||
def OnRtnTrade(self, pTrade):
|
||
print("||成交回报||", pTrade)
|
||
|
||
def OnRspOrderInsert(self, pInputOrder, pRspInfo, nRequestID, bIsLast):
|
||
print("||OnRspOrderInsert||", pInputOrder, pRspInfo, nRequestID, bIsLast)
|
||
|
||
# 订单状态通知
|
||
def OnRtnOrder(self, pOrder):
|
||
print("||订单回报||", pOrder)
|
||
|
||
def cal_sig(self, symbol_queue):
|
||
while True:
|
||
try:
|
||
data = symbol_queue.get(
|
||
block=True, timeout=5
|
||
) # 如果5秒没收到新的tick行情，则抛出异常
|
||
instrument_id = data["InstrumentID"].decode() # 品种代码
|
||
size = symbol_queue.qsize()
|
||
if size > 1:
|
||
print(
|
||
f"当前{instrument_id}共享队列长度为{size}, 有点阻塞！！！！！"
|
||
)
|
||
self.read_to_csv(instrument_id)
|
||
self.day_data_reset(instrument_id)
|
||
param = self.param_dict[instrument_id]
|
||
self.品种 = instrument_id
|
||
self.tickcome(data)
|
||
trade_df = trade_dfs[instrument_id]
|
||
# 新K线开始，启动交易程序 and 保存行情数据
|
||
self.read_to_csv(instrument_id)
|
||
if len(trade_df) > param.cont_df:
|
||
# 检查文件是否存在
|
||
csv_file_path = f"traderdata/{instrument_id}_ofdata.csv"
|
||
# if os.path.exists(csv_file_path):
|
||
# #jerome :保存数增加'delta累计'、POC、、终极平滑值、趋势方向
|
||
# # 仅保存最后一行数据
|
||
# trade_df.tail(1).to_csv(
|
||
# csv_file_path, mode="a", header=False, index=False
|
||
# )
|
||
# else:
|
||
# # 创建新文件并保存整个DataFrame
|
||
# trade_df.to_csv(csv_file_path, index=False)
|
||
# 检查是否存在重复行
|
||
if os.path.exists(csv_file_path):
|
||
existing_df = pd.read_csv(csv_file_path, usecols=range(12))
|
||
# 获取要写入的新数据
|
||
new_data = trade_df.tail(1)
|
||
|
||
# 检查新数据是否与现有数据重复
|
||
is_duplicate = False
|
||
for _, row in existing_df.iterrows():
|
||
if (row['datetime'] == new_data['datetime'].iloc[0] and
|
||
row['price'] == new_data['price'].iloc[0] and
|
||
row['Ask'] == new_data['Ask'].iloc[0] and
|
||
row['Bid'] == new_data['Bid'].iloc[0] and
|
||
row['symbol'] == new_data['symbol'].iloc[0] and
|
||
row['delta'] == new_data['delta'].iloc[0] and
|
||
row['close'] == new_data['close'].iloc[0] and
|
||
row['open'] == new_data['open'].iloc[0] and
|
||
row['high'] == new_data['high'].iloc[0] and
|
||
row['low'] == new_data['low'].iloc[0] and
|
||
row['volume'] == new_data['volume'].iloc[0] and
|
||
row['dj'] == new_data['dj'].iloc[0]):
|
||
is_duplicate = True
|
||
break
|
||
|
||
# 检查Ask和Bid的值是否为空或全为0
|
||
ask_value = new_data['Ask'].iloc[0]
|
||
bid_value = new_data['Bid'].iloc[0]
|
||
is_valid_data = (
|
||
ask_value != [] and
|
||
ask_value != [0] and
|
||
bid_value != [] and
|
||
bid_value != [0]
|
||
)
|
||
|
||
if not is_duplicate and is_valid_data:
|
||
# 如果没有重复且数据有效，则写入新数据
|
||
new_data.to_csv(
|
||
csv_file_path, mode="a", header=False, index=False
|
||
)
|
||
else:
|
||
# 如果文件不存在，直接写入新数据
|
||
trade_df.to_csv(csv_file_path, index=False)
|
||
|
||
# 更新跟踪止损价格
|
||
if param.long_trailing_stop_price > 0 and param.pos > 0:
|
||
|
||
param.long_trailing_stop_price = (
|
||
trade_df["low"].iloc[-1]
|
||
if param.long_trailing_stop_price < trade_df["low"].iloc[-1]
|
||
else param.long_trailing_stop_price
|
||
)
|
||
self.save_to_csv(instrument_id)
|
||
|
||
if param.short_trailing_stop_price > 0 and param.pos < 0:
|
||
|
||
param.short_trailing_stop_price = (
|
||
trade_df["high"].iloc[-1]
|
||
if trade_df["high"].iloc[-1]
|
||
< param.short_trailing_stop_price
|
||
else param.short_trailing_stop_price
|
||
)
|
||
self.save_to_csv(instrument_id)
|
||
|
||
param.out_long = param.long_trailing_stop_price * (
|
||
1 - param.trailing_stop_percent
|
||
)
|
||
param.out_short = param.short_trailing_stop_price * (
|
||
1 + param.trailing_stop_percent
|
||
)
|
||
# 跟踪出场
|
||
if param.out_long > 0:
|
||
print(
|
||
"datetime+sig: ",
|
||
trade_df["datetime"].iloc[-1],
|
||
"预设——多头止盈——",
|
||
"TR",
|
||
param.out_long,
|
||
"low",
|
||
trade_df["low"].iloc[-1],
|
||
)
|
||
if (
|
||
trade_df["low"].iloc[-1] < param.out_long
|
||
and param.pos > 0
|
||
and param.sl_long_price > 0
|
||
and trade_df["low"].iloc[-1] > param.sl_long_price
|
||
):
|
||
print(
|
||
"datetime+sig: ",
|
||
trade_df["datetime"].iloc[-1],
|
||
"多头止盈",
|
||
"TR",
|
||
param.out_long,
|
||
"low",
|
||
trade_df["low"].iloc[-1],
|
||
)
|
||
# 平多
|
||
self.insert_order(
|
||
data["ExchangeID"],
|
||
data["InstrumentID"],
|
||
data["BidPrice1"] - param.py,
|
||
param.Lots,
|
||
b"1",
|
||
b"1",
|
||
)
|
||
self.insert_order(
|
||
data["ExchangeID"],
|
||
data["InstrumentID"],
|
||
data["BidPrice1"] - param.py,
|
||
param.Lots,
|
||
b"1",
|
||
b"3",
|
||
)
|
||
param.long_trailing_stop_price = 0
|
||
param.out_long = 0
|
||
param.sl_long_price = 0
|
||
param.pos = 0
|
||
self.save_to_csv(instrument_id)
|
||
|
||
if param.out_short > 0:
|
||
print(
|
||
"datetime+sig: ",
|
||
trade_df["datetime"].iloc[-1],
|
||
"预设——空头止盈——: ",
|
||
"TR",
|
||
param.out_short,
|
||
"high",
|
||
trade_df["high"].iloc[-1],
|
||
)
|
||
if (
|
||
trade_df["high"].iloc[-1] > param.out_short
|
||
and param.pos < 0
|
||
and param.sl_shor_price > 0
|
||
and trade_df["high"].iloc[-1] < param.sl_shor_price
|
||
):
|
||
print(
|
||
"datetime+sig: ",
|
||
trade_df["datetime"].iloc[-1],
|
||
"空头止盈: ",
|
||
"TR",
|
||
param.out_short,
|
||
"high",
|
||
trade_df["high"].iloc[-1],
|
||
)
|
||
# 平空
|
||
self.insert_order(
|
||
data["ExchangeID"],
|
||
data["InstrumentID"],
|
||
data["AskPrice1"] + param.py,
|
||
param.Lots,
|
||
b"0",
|
||
b"1",
|
||
)
|
||
self.insert_order(
|
||
data["ExchangeID"],
|
||
data["InstrumentID"],
|
||
data["AskPrice1"] + param.py,
|
||
param.Lots,
|
||
b"0",
|
||
b"3",
|
||
)
|
||
param.short_trailing_stop_price = 0
|
||
param.sl_shor_price = 0
|
||
self.out_shor = 0
|
||
param.pos = 0
|
||
self.save_to_csv(instrument_id)
|
||
|
||
# 固定止损
|
||
fixed_stop_loss_L = param.sl_long_price * (
|
||
1 - param.fixed_stop_loss_percent
|
||
)
|
||
if param.pos > 0:
|
||
print(
|
||
"datetime+sig: ",
|
||
trade_df["datetime"].iloc[-1],
|
||
"预设——多头止损",
|
||
"SL",
|
||
fixed_stop_loss_L,
|
||
"close",
|
||
trade_df["close"].iloc[-1],
|
||
)
|
||
if (
|
||
param.sl_long_price > 0
|
||
and fixed_stop_loss_L > 0
|
||
and param.pos > 0
|
||
and trade_df["close"].iloc[-1] < fixed_stop_loss_L
|
||
):
|
||
print(
|
||
"datetime+sig: ",
|
||
trade_df["datetime"].iloc[-1],
|
||
"多头止损",
|
||
"SL",
|
||
fixed_stop_loss_L,
|
||
"close",
|
||
trade_df["close"].iloc[-1],
|
||
)
|
||
# 平多
|
||
self.insert_order(
|
||
data["ExchangeID"],
|
||
data["InstrumentID"],
|
||
data["BidPrice1"] - param.py,
|
||
param.Lots,
|
||
b"1",
|
||
b"1",
|
||
)
|
||
self.insert_order(
|
||
data["ExchangeID"],
|
||
data["InstrumentID"],
|
||
data["BidPrice1"] - param.py,
|
||
param.Lots,
|
||
b"1",
|
||
b"3",
|
||
)
|
||
param.long_trailing_stop_price = 0
|
||
param.sl_long_price = 0
|
||
param.out_long = 0
|
||
param.pos = 0
|
||
self.save_to_csv(instrument_id)
|
||
|
||
fixed_stop_loss_S = param.sl_shor_price * (
|
||
1 + param.fixed_stop_loss_percent
|
||
)
|
||
if param.pos < 0:
|
||
print(
|
||
"datetime+sig: ",
|
||
trade_df["datetime"].iloc[-1],
|
||
"预设——空头止损",
|
||
"SL",
|
||
fixed_stop_loss_S,
|
||
"close",
|
||
trade_df["close"].iloc[-1],
|
||
)
|
||
if (
|
||
param.sl_shor_price > 0
|
||
and fixed_stop_loss_S > 0
|
||
and param.pos < 0
|
||
and trade_df["close"].iloc[-1] > fixed_stop_loss_S
|
||
):
|
||
print(
|
||
"datetime+sig: ",
|
||
trade_df["datetime"].iloc[-1],
|
||
"空头止损",
|
||
"SL",
|
||
fixed_stop_loss_S,
|
||
"close",
|
||
trade_df["close"].iloc[-1],
|
||
)
|
||
# 平空
|
||
self.insert_order(
|
||
data["ExchangeID"],
|
||
data["InstrumentID"],
|
||
data["AskPrice1"] + param.py,
|
||
param.Lots,
|
||
b"0",
|
||
b"1",
|
||
)
|
||
self.insert_order(
|
||
data["ExchangeID"],
|
||
data["InstrumentID"],
|
||
data["AskPrice1"] + param.py,
|
||
param.Lots,
|
||
b"0",
|
||
b"3",
|
||
)
|
||
param.short_trailing_stop_price = 0
|
||
param.sl_shor_price = 0
|
||
param.out_short = 0
|
||
param.pos = 0
|
||
self.save_to_csv(instrument_id)
|
||
|
||
# 日均线
|
||
# AROONOSC ：https://zhuanlan.zhihu.com/p/645010879
|
||
# if len(trade_df["close"]) >= 120:
|
||
# trade_df["dayma"] = trade_df["close"][-120:].mean()
|
||
# print("trade_df长度:", len(trade_df["close"]))
|
||
# print("120条之上的dayma的值:", trade_df["dayma"])
|
||
# else:
|
||
# trade_df["dayma"] = trade_df["close"].mean()
|
||
# print("120条之下的dayma的值:", trade_df["dayma"])
|
||
# print("trade_df长度:", len(trade_df["close"]))
|
||
day_df = {}
|
||
day_df = futures_main_day(
|
||
instrument_id, 20
|
||
) # futures_main_day(trade_df["symbol"], 20)
|
||
day_df["5day_ma"] = day_df["收盘价"].rolling(window=5).mean()
|
||
day_df["5day_ma"].iloc[-1]
|
||
# trade_df["aroon_osc"] = tb.AROONOSC(trade_df["high"], trade_df["low"], 5)
|
||
# trade_df["rinei_T3"] = tb.T3(np.array(trade_df["dayma"]))
|
||
# print("交易品种为:", instrument_id)
|
||
# print("昨日5日均线:", day_df["5day_ma"].iloc[-1])
|
||
# print("昨日收盘价:", day_df["收盘价"].iloc[-1])
|
||
|
||
# 计算累积的delta值datetime.strptime(str_time, "%Y-%m-%d %H:%M:%S")
|
||
trade_df["delta"] = trade_df["delta"].astype(float)
|
||
trade_df['datetime'] = pd.to_datetime(trade_df['datetime'], format='mixed')
|
||
trade_df['delta累计'] = trade_df.groupby(trade_df['datetime'].dt.date)['delta'].cumsum()
|
||
# if len(trade_df['datetime']) == 1:
|
||
# trade_df["delta累计"].iloc[-1] = trade_df["delta"].iloc[-1]
|
||
# else:
|
||
# trade_time_1 = datetime.strptime(trade_df['datetime'].iloc[-1], "%Y-%m-%d %H:%M:%S")
|
||
# trade_time_2 = datetime.strptime(trade_df['datetime'].iloc[-2], "%Y-%m-%d %H:%M:%S")
|
||
|
||
# if trade_time_1.date() != trade_time_2.date():
|
||
# trade_df["delta累计"].iloc[-1] = trade_df["delta"].iloc[-1]
|
||
# else:
|
||
# trade_df["delta累计"].iloc[-1] = trade_df["delta"].iloc[-1]+trade_df["delta"].iloc[-2]
|
||
# trade_df["delta累计"] = trade_df["delta"].cumsum()
|
||
|
||
# 获取第三大值和第三小值
|
||
abs_delta = max(trade_df["delta"].iloc[-120:-1], default=0) - min(
|
||
trade_df["delta"].iloc[-120:-1], default=0
|
||
)
|
||
print("abs_delta:", abs_delta)
|
||
# third_largest_delta = np.sort(arr_delta)[-2]
|
||
# third_smallest_delta = np.sort(arr_delta)[2]
|
||
|
||
abs_delta累计 = max(
|
||
trade_df["delta累计"].iloc[-120:-1], default=0
|
||
) - min(trade_df["delta累计"].iloc[-120:-1], default=0)
|
||
print("abs_delta累计:", abs_delta累计)
|
||
# third_largest_delta累计 = np.sort(arr_delta累计)[-2]
|
||
# third_smallest_delta累计 = np.sort(arr_delta累计)[2]
|
||
|
||
# 大于日均线
|
||
# 开多1 = trade_df["dayma"].iloc[-1] > 0 and trade_df["close"].iloc[-1] > trade_df["dayma"].iloc[-1]
|
||
# 开多1 = trade_df["aroon_osc"].iloc[-1] > 0
|
||
# 开多1 = trade_df["close"].iloc[-1] > trade_df[
|
||
# "rinei_T3"].iloc[-1]
|
||
开多1 = day_df["收盘价"].iloc[-1] > day_df["5day_ma"].iloc[-1]
|
||
|
||
# 累计多空净量大于X
|
||
# 开多4 = (
|
||
# trade_df["delta累计"].iloc[-1] > param.sum_delta and trade_df["delta"].iloc[-1] > param.delta
|
||
# )
|
||
开多4 = trade_df["delta"].iloc[-1] > param.delta and trade_df["delta累计"].iloc[-1] > param.sum_delta
|
||
# 开多4 = trade_df["delta累计"].iloc[-1] > np.sort(trade_df["delta累计"].iloc[-120:-1], default=0)[-2] and trade_df["delta"].iloc[-1] > np.sort(trade_df["delta"].iloc[-120:-1], default=0)[-2]
|
||
# 开多4 = trade_df["delta累计"].iloc[-1] > third_largest_delta累计 and trade_df["delta"].iloc[-1] > third_largest_delta
|
||
|
||
# 小于日均线
|
||
# 开空1 = trade_df["dayma"].iloc[-1] > 0 and trade_df["close"].iloc[-1] < trade_df["dayma"].iloc[-1]
|
||
# 开空1 = trade_df["aroon_osc"].iloc[-1] < 0
|
||
# 开空1 = trade_df["close"].iloc[-1] < trade_df[
|
||
# "rinei_T3"].iloc[-1]
|
||
开空1 = day_df["收盘价"].iloc[-1] < day_df["5day_ma"].iloc[-1]
|
||
|
||
# 累计多空净量小于X
|
||
开空4 = trade_df["delta"].iloc[-1] < -param.delta and trade_df["delta累计"].iloc[-1] <- param.sum_delta
|
||
|
||
# 开空4 = trade_df["delta累计"].iloc[-1] < np.sort(trade_df["delta累计"].iloc[-120:-1], default=0)[2] and trade_df["delta"].iloc[-1] < np.sort(trade_df["delta"].iloc[-120:-1], default=0)[2]
|
||
# 开空4 = trade_df["delta累计"].iloc[-1] < third_smallest_delta累计 and trade_df["delta"].iloc[-1] < third_smallest_delta
|
||
开多组合 = (
|
||
# 开多1
|
||
开多4
|
||
and trade_df["dj"].iloc[-1] > param.dj_X
|
||
and datetime.now().time() < s_time(14,55)
|
||
# and len(trade_df) > 120
|
||
)
|
||
开空条件 = (
|
||
# 开空1
|
||
开空4
|
||
and trade_df["dj"].iloc[-1] < -param.dj_X
|
||
and datetime.now().time() < s_time(14,55)
|
||
# and len(trade_df) > 120
|
||
)
|
||
|
||
平多条件 = (trade_df["dj"].iloc[-1] < -param.dj_X) or (datetime.now().time() >= s_time(14,55))
|
||
平空条件 = (trade_df["dj"].iloc[-1] > param.dj_X) or (datetime.now().time() >= s_time(14,55))
|
||
# 开仓
|
||
# 多头开仓条件
|
||
if param.pos < 0 and 平空条件:
|
||
print(
|
||
"平空： ",
|
||
"ExchangeID: ",
|
||
data["ExchangeID"],
|
||
"InstrumentID",
|
||
data["InstrumentID"],
|
||
"AskPrice1",
|
||
data["AskPrice1"] + param.py,
|
||
)
|
||
# 平空
|
||
self.insert_order(
|
||
data["ExchangeID"],
|
||
data["InstrumentID"],
|
||
data["AskPrice1"] + param.py,
|
||
param.Lots,
|
||
b"0",
|
||
b"1",
|
||
)
|
||
self.insert_order(
|
||
data["ExchangeID"],
|
||
data["InstrumentID"],
|
||
data["AskPrice1"] + param.py,
|
||
param.Lots,
|
||
b"0",
|
||
b"3",
|
||
)
|
||
|
||
param.pos = 0
|
||
param.sl_shor_price = 0
|
||
param.short_trailing_stop_price = 0
|
||
print(
|
||
"datetime+sig: ",
|
||
trade_df["datetime"].iloc[-1],
|
||
"反手平空：",
|
||
"平仓价格：",
|
||
data["AskPrice1"] + param.py,
|
||
"堆积数：",
|
||
trade_df["dj"].iloc[-1],
|
||
)
|
||
self.save_to_csv(instrument_id)
|
||
|
||
# 发送邮件
|
||
# text = f"平空交易： 交易品种为{data['InstrumentID']}, 交易时间为{trade_df['datetime'].iloc[-1]}, 反手平空的平仓价格为{data['AskPrice1']+param.py}, 交易手数位{param.Lots}"
|
||
text = f"C_S_T: ID:{data['InstrumentID']}, datetime:{trade_df['datetime'].iloc[-1]}, C_S_T_Price:{data['AskPrice1'] + param.py}, T_Lots:{param.Lots}"
|
||
send_feishu_message(text)
|
||
|
||
if param.pos == 0 and 开多组合:
|
||
print(
|
||
"开多： ",
|
||
"ExchangeID: ",
|
||
data["ExchangeID"],
|
||
"InstrumentID",
|
||
data["InstrumentID"],
|
||
"AskPrice1",
|
||
data["AskPrice1"] + param.py,
|
||
)
|
||
# 开多
|
||
self.insert_order(
|
||
data["ExchangeID"],
|
||
data["InstrumentID"],
|
||
data["AskPrice1"] + param.py,
|
||
param.Lots,
|
||
b"0",
|
||
b"0",
|
||
)
|
||
print(
|
||
"datetime+sig: ",
|
||
trade_df["datetime"].iloc[-1],
|
||
"多头开仓",
|
||
"开仓价格：",
|
||
data["AskPrice1"] + param.py,
|
||
"堆积数：",
|
||
trade_df["dj"].iloc[-1],
|
||
)
|
||
param.pos = 1
|
||
param.long_trailing_stop_price = data["AskPrice1"]
|
||
param.sl_long_price = data["AskPrice1"]
|
||
self.save_to_csv(instrument_id)
|
||
|
||
# 发送邮件
|
||
text = f"O_L_T ID:{data['InstrumentID']}, datetime:{trade_df['datetime'].iloc[-1]}, O_L_T_Price:{data['AskPrice1'] + param.py}, T_Lots:{param.Lots}"
|
||
send_feishu_message(text)
|
||
|
||
if param.pos > 0 and 平多条件:
|
||
print(
|
||
"平多： ",
|
||
"ExchangeID: ",
|
||
data["ExchangeID"],
|
||
"InstrumentID",
|
||
data["InstrumentID"],
|
||
"BidPrice1",
|
||
data["BidPrice1"] - param.py,
|
||
)
|
||
# 平多
|
||
self.insert_order(
|
||
data["ExchangeID"],
|
||
data["InstrumentID"],
|
||
data["BidPrice1"] - param.py,
|
||
param.Lots,
|
||
b"1",
|
||
b"1",
|
||
)
|
||
self.insert_order(
|
||
data["ExchangeID"],
|
||
data["InstrumentID"],
|
||
data["BidPrice1"] - param.py,
|
||
param.Lots,
|
||
b"1",
|
||
b"3",
|
||
)
|
||
|
||
param.pos = 0
|
||
param.long_trailing_stop_price = 0
|
||
param.sl_long_price = 0
|
||
print(
|
||
"datetime+sig: ",
|
||
trade_df["datetime"].iloc[-1],
|
||
"反手平多",
|
||
"平仓价格：",
|
||
data["BidPrice1"] - param.py,
|
||
"堆积数：",
|
||
trade_df["dj"].iloc[-1],
|
||
)
|
||
self.save_to_csv(instrument_id)
|
||
|
||
# 发送邮件
|
||
# text = f"平多交易： 交易品种为{data['InstrumentID']}, 交易时间为{trade_df['datetime'].iloc[-1]}, 反手平多的平仓价格{data['BidPrice1']-param.py}, 交易手数位{param.Lots}"
|
||
text = f"C_L_T: ID:{data['InstrumentID']}, datetime:{trade_df['datetime'].iloc[-1]}, C_L_T_Price:{data['BidPrice1'] - param.py}, T_Lots:{param.Lots}"
|
||
send_feishu_message(text)
|
||
|
||
if param.pos == 0 and 开空条件:
|
||
print(
|
||
"开空： ",
|
||
"ExchangeID: ",
|
||
data["ExchangeID"],
|
||
"InstrumentID",
|
||
data["InstrumentID"],
|
||
"BidPrice1",
|
||
data["BidPrice1"],
|
||
)
|
||
# 开空
|
||
self.insert_order(
|
||
data["ExchangeID"],
|
||
data["InstrumentID"],
|
||
data["BidPrice1"] - param.py,
|
||
param.Lots,
|
||
b"1",
|
||
b"0",
|
||
)
|
||
print(
|
||
"datetime+sig: ",
|
||
trade_df["datetime"].iloc[-1],
|
||
"空头开仓",
|
||
"开仓价格：",
|
||
data["BidPrice1"] - param.py,
|
||
"堆积数：",
|
||
trade_df["dj"].iloc[-1],
|
||
)
|
||
param.pos = -1
|
||
param.short_trailing_stop_price = data["BidPrice1"]
|
||
param.sl_shor_price = data["BidPrice1"]
|
||
self.save_to_csv(instrument_id)
|
||
|
||
# 发送邮件
|
||
text = f"O_S_T: ID:{data['InstrumentID']}, datetime:{trade_df['datetime'].iloc[-1]}, O_S_T_Price:{data['BidPrice1'] - param.py}, T_Lots:{param.Lots}"
|
||
send_feishu_message(text)
|
||
|
||
print(trade_df)
|
||
print("------------------------------------------------")
|
||
# print(trade_df.iloc[0])
|
||
# print(trade_df.iloc[-1])
|
||
param.cont_df = len(trade_df)
|
||
except queue.Empty:
|
||
# print(f"当前合约队列为空，等待新数据插入。")
|
||
pass
|
||
|
||
# 将CTP推送的行情数据分发给对应线程队列去执行
|
||
def distribute_tick(self):
|
||
while True:
|
||
if self.status == 0:
|
||
data = None
|
||
while not self.md_queue.empty():
|
||
data = self.md_queue.get(block=False)
|
||
instrument_id = data["InstrumentID"].decode() # 品种代码
|
||
try:
|
||
self.queue_dict[instrument_id].put(
|
||
data, block=False
|
||
) # 往对应合约队列中插入行情
|
||
# print(f"{instrument_id}合约数据插入。")
|
||
except queue.Full:
|
||
# 当某个线程阻塞导致对应队列容量超限时抛出异常，不会影响其他合约的信号计算
|
||
print(
|
||
f"{instrument_id}合约信号计算阻塞导致对应队列已满，请检查对应代码逻辑后重启。"
|
||
)
|
||
else:
|
||
time.sleep(1)
|
||
|
||
def start(self, param_dict):
|
||
threads = []
|
||
self.param_dict = param_dict
|
||
|
||
for symbol in param_dict.keys():
|
||
# folder_path = "traderdata"
|
||
# ofdata_file_path = os.path.join("traderdata", f"{str(symbol)}_ofdata.csv")
|
||
if os.path.exists(f"traderdata/{symbol}_ofdata.csv"):
|
||
columns = [
|
||
"price",
|
||
"Ask",
|
||
"Bid",
|
||
"symbol",
|
||
"datetime",
|
||
"delta",
|
||
"close",
|
||
"open",
|
||
"high",
|
||
"low",
|
||
"volume",
|
||
"dj",
|
||
]
|
||
# import csv
|
||
# with open(f"traderdata/{symbol}_ofdata.csv", "r") as f:
|
||
# reader = csv.reader(f)
|
||
# for i, row in enumerate(reader, 1):
|
||
# if len(row) != 12:
|
||
# print(f"Line {i} has {len(row)} columns: {row}")
|
||
trade_dfs[symbol] = pd.read_csv(
|
||
f"traderdata/{symbol}_ofdata.csv", usecols=columns
|
||
)
|
||
|
||
else:
|
||
trade_dfs[symbol] = pd.DataFrame({})
|
||
self.queue_dict[symbol] = queue.Queue(
|
||
20
|
||
) # 为每个合约创建一个限制数为10的队列，当计算发生阻塞导致队列达到限制数时会抛出异常
|
||
t = threading.Thread(
|
||
target=self.cal_sig, args=(self.queue_dict[symbol],)
|
||
) # 为每个合约单独创建一个线程计算开仓逻辑
|
||
threads.append(t)
|
||
t.start()
|
||
self.distribute_tick()
|
||
for t in threads:
|
||
t.join()
|
||
|
||
|
||
def run_trader(
|
||
param_dict,
|
||
broker_id,
|
||
td_server,
|
||
investor_id,
|
||
password,
|
||
app_id,
|
||
auth_code,
|
||
md_queue=None,
|
||
page_dir="",
|
||
private_resume_type=2,
|
||
public_resume_type=2,
|
||
):
|
||
my_trader = MyTrader(
|
||
broker_id,
|
||
td_server,
|
||
investor_id,
|
||
password,
|
||
app_id,
|
||
auth_code,
|
||
md_queue,
|
||
page_dir,
|
||
private_resume_type,
|
||
public_resume_type,
|
||
)
|
||
my_trader.start(param_dict)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 注意：运行前请先安装好algoplus,
|
||
# pip install AlgoPlus
|
||
# http://www.algo.plus/ctp/python/0103001.html
|
||
|
||
param_dict = {}
|
||
param_dict["IM2504"] = ParamObj(
|
||
symbol="IM2504",
|
||
Lots=1,
|
||
py=5,
|
||
trailing_stop_percent=0.01,
|
||
fixed_stop_loss_percent=0.02,
|
||
dj_X=8,
|
||
delta=500,
|
||
sum_delta=800,
|
||
失衡=3,
|
||
堆积=3,
|
||
周期="2min",
|
||
)
|
||
# param_dict["IF2504"] = ParamObj(
|
||
# symbol="IF2504",
|
||
# Lots=1,
|
||
# py=5,
|
||
# trailing_stop_percent=0.01,
|
||
# fixed_stop_loss_percent=0.02,
|
||
# dj_X=5,
|
||
# delta=300,
|
||
# sum_delta=300,
|
||
# 失衡=3,
|
||
# 堆积=3,
|
||
# 周期="2min",
|
||
# )
|
||
# param_dict["IH2504"] = ParamObj(
|
||
# symbol="IH2504",
|
||
# Lots=1,
|
||
# py=5,
|
||
# trailing_stop_percent=0.01,
|
||
# fixed_stop_loss_percent=0.02,
|
||
# dj_X=5,
|
||
# delta=300,
|
||
# sum_delta=300,
|
||
# 失衡=3,
|
||
# 堆积=3,
|
||
# 周期="2min",
|
||
# )
|
||
param_dict["lh2505"] = ParamObj(
|
||
symbol="lh2505",
|
||
Lots=1,
|
||
py=5,
|
||
trailing_stop_percent=0.01,
|
||
fixed_stop_loss_percent=0.02,
|
||
dj_X=8,
|
||
delta=1500,
|
||
sum_delta=2000,
|
||
失衡=3,
|
||
堆积=3,
|
||
周期="1min",
|
||
)
|
||
param_dict["ag2505"] = ParamObj(
|
||
symbol="ag2505",
|
||
Lots=1,
|
||
py=5,
|
||
trailing_stop_percent=0.01,
|
||
fixed_stop_loss_percent=0.02,
|
||
dj_X=8,
|
||
delta=1500,
|
||
sum_delta=2000,
|
||
失衡=3,
|
||
堆积=3,
|
||
周期="2min",
|
||
)
|
||
param_dict["ni2505"] = ParamObj(
|
||
symbol="ni2505",
|
||
Lots=1,
|
||
py=5,
|
||
trailing_stop_percent=0.01,
|
||
fixed_stop_loss_percent=0.02,
|
||
dj_X=8,
|
||
delta=1500,
|
||
sum_delta=2000,
|
||
失衡=3,
|
||
堆积=3,
|
||
周期="2min",
|
||
)
|
||
# 用simnow模拟，不要忘记屏蔽下方实盘的future_account字典
|
||
# SIMULATE_SERVER = {
|
||
# '电信1': {'BrokerID': 9999, 'TDServer': "180.168.146.187:10201", 'MDServer': '180.168.146.187:10211', 'AppID': 'simnow_client_test', 'AuthCode': '0000000000000000'},
|
||
# '电信2': {'BrokerID': 9999, 'TDServer': "180.168.146.187:10202", 'MDServer': '180.168.146.187:10212', 'AppID': 'simnow_client_test', 'AuthCode': '0000000000000000'},
|
||
# '移动': {'BrokerID': 9999, 'TDServer': "218.202.237.33:10203", 'MDServer': '218.202.237.33:10213', 'AppID': 'simnow_client_test', 'AuthCode': '0000000000000000'},
|
||
# 'TEST': {'BrokerID': 9999, 'TDServer': "180.168.146.187:10130", 'MDServer': '180.168.146.187:10131', 'AppID': 'simnow_client_test', 'AuthCode': '0000000000000000'},
|
||
# 'N视界': {'BrokerID': 10010, 'TDServer': "210.14.72.12:4600", 'MDServer': '210.14.72.12:4602', 'AppID': '', 'AuthCode': ''},
|
||
# }
|
||
# BrokerID统一为：9999
|
||
# 支持上期所期权、能源中心期权、中金所期权、广期所期权、郑商所期权、大商所期权
|
||
# 第一组
|
||
# Trade Front：180.168.146.187:10201，Market Front：180.168.146.187:10211；【电信】（看穿式前置，使用监控中心生产秘钥）
|
||
|
||
# 第二组
|
||
# Trade Front：180.168.146.187:10202，Market Front：180.168.146.187:10212；【电信】（看穿式前置，使用监控中心生产秘钥）
|
||
|
||
# 第三组
|
||
# Trade Front：218.202.237.33:10203，Market Front：218.202.237.33:10213；【移动】（看穿式前置，使用监控中心生产秘钥）
|
||
|
||
# 用户注册后，默认的APPID为simnow_client_test，认证码为0000000000000000（16个0），默认开启终端认证，程序化用户可以选择不开终端认证接入。
|
||
|
||
future_account = get_simulate_account(
|
||
investor_id="223828", # simnow账户，注意是登录账户的ID，SIMNOW个人首页查看
|
||
password="Zj1234!@#%", # simnow密码
|
||
server_name="电信1", # 电信1、电信2、移动、TEST、N视界
|
||
subscribe_list=list(param_dict.keys()), # 合约列表
|
||
)
|
||
|
||
# 实盘用这个，不要忘记屏蔽上方simnow的future_account字典
|
||
# future_account = FutureAccount(
|
||
# broker_id='9999', # 期货公司BrokerID
|
||
# server_dict={'TDServer': "180.168.146.187:10201", 'MDServer': '180.168.146.187:10211'}, # TDServer为交易服务器，MDServer为行情服务器。服务器地址格式为"ip:port。"
|
||
# reserve_server_dict={}, # 备用服务器地址
|
||
# investor_id='223828', # 账户
|
||
# password='Zj1234!@#%', # 密码
|
||
# app_id='simnow_client_test', # 认证使用AppID
|
||
# auth_code='0000000000000000', # 认证使用授权码
|
||
# subscribe_list=list(param_dict.keys()), # 订阅合约列表
|
||
# md_flow_path='./log', # MdApi流文件存储地址，默认MD_LOCATION
|
||
# td_flow_path='./log', # TraderApi流文件存储地址，默认TD_LOCATION
|
||
# )
|
||
|
||
# 实盘用这个，不要忘记屏蔽上方simnow的future_account字典
|
||
# future_account = FutureAccount(
|
||
# broker_id='8888', # 期货公司BrokerID
|
||
# server_dict={'TDServer': "103.140.14.210:43205", 'MDServer': '103.140.14.210:43173'}, # TDServer为交易服务器，MDServer为行情服务器。服务器地址格式为"ip:port。"
|
||
# reserve_server_dict={}, # 备用服务器地址
|
||
# investor_id='155878', # 账户
|
||
# password='Zj82334475', # 密码
|
||
# app_id='vntech_vnpy_2.0', # 认证使用AppID
|
||
# auth_code='N46EKN6TJ9U7V06V', # 认证使用授权码
|
||
# subscribe_list=list(param_dict.keys()), # 订阅合约列表
|
||
# md_flow_path='./log', # MdApi流文件存储地址，默认MD_LOCATION
|
||
# td_flow_path='./log', # TraderApi流文件存储地址，默认TD_LOCATION
|
||
# )
|
||
|
||
print("开始", len(future_account.subscribe_list))
|
||
# 共享队列
|
||
share_queue = Queue(maxsize=200)
|
||
|
||
# 行情进程
|
||
md_process = Process(target=run_tick_engine, args=(future_account, [share_queue]))
|
||
|
||
# 交易进程
|
||
trader_process = Process(
|
||
target=run_trader,
|
||
args=(
|
||
param_dict,
|
||
future_account.broker_id,
|
||
future_account.server_dict["TDServer"],
|
||
future_account.investor_id,
|
||
future_account.password,
|
||
future_account.app_id,
|
||
future_account.auth_code,
|
||
share_queue, # 队列
|
||
future_account.td_flow_path,
|
||
),
|
||
)
|
||
|
||
md_process.start()
|
||
trader_process.start()
|
||
|
||
md_process.join()
|
||
trader_process.join()
|