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()