"""
该代码的主要目的是处理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 AlgoPlus.CTP.MdApi import run_tick_engine
# from AlgoPlus.CTP.FutureAccount import get_simulate_account
# from AlgoPlus.CTP.FutureAccount import FutureAccount
# from AlgoPlus.CTP.TraderApiBase import TraderApiBase

from CtpPlus.CTP.MdApi import run_tick_engine
from CtpPlus.CTP.FutureAccount import get_simulate_account
from CtpPlus.CTP.FutureAccount import FutureAccount
from CtpPlus.CTP.TraderApiBase import TraderApiBase

# 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 ast

# 加入邮件通知
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 = {}


time_period = 30
delta_rate = 0.8
dj_rate = 0.8

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),
}

fees_df = pd.read_csv('./futures_fees_info.csv', header = 0, usecols= [1, 3, 5, 13, 15],names=['合约代码', '品种代码', '合约乘数', '做多保证金率（按金额）', '做空保证金率（按金额）'])
contacts_df = pd.read_csv('./main_contacts.csv', header = 0, usecols= [16, 17],names=['主连代码', '品种代码'])  

def get_main_contact_on_time(main_symbol_code,contacts_df):
    main_symbol = contacts_df[contacts_df['品种代码'] == main_symbol_code]['主连代码'].iloc[0]
    # print("最终使用的主连代码：",main_symbol)
    return main_symbol#.encode('ascii')

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 get_otm_put_strike_price(option_finance_board_df, future_price):
#     # 计算距离当前期货价格最近的行权价
#     option_finance_board_df['strike_diff'] = abs(option_finance_board_df['行权价'] - future_price)
#     closest_row = option_finance_board_df.loc[option_finance_board_df['strike_diff'].idxmin()]
#     otm_put_strike_price = closest_row['行权价']
#     return otm_put_strike_price


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) #AI策略飞书地址
        response = requests.post("https://open.feishu.cn/open-apis/bot/v2/hook/fae322eb-1ff7-4133-ba00-0ca4895d205e", 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(TraderApiBase):

    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"),
            # "created_at": datetime.strptime(created_at, "-- %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)


                    # 计算累积的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['datetime'] = pd.to_datetime(trade_df['datetime'], format='%Y-%m-%d %H:%M:%S')

                    # 自定义分组逻辑：前一日21:00至当日15:00为一天
                    def get_trading_day(dt):
                        # 如果时间在21:00之后，属于下一个交易日
                        if dt.hour >= 21:
                            return (dt + pd.Timedelta(days=1)).date()
                        # 如果时间在15:00之前，属于当前交易日
                        elif dt.hour < 15:
                            return dt.date()
                        # 15:00-21:00之间的数据属于当前交易日
                        else:
                            return dt.date()

                    trade_df['trading_day'] = trade_df['datetime'].apply(get_trading_day)
                    
                    # 将日期转换为字符串
                    trade_df['trading_day'] = trade_df['trading_day'].astype(str)  

                    # 按交易日计算delta累计
                    trade_df['delta累计'] = trade_df.groupby('trading_day')['delta'].cumsum()

                    trade_df = trade_df.fillna('缺值')#fillna(value=0)

                    def ultimate_smoother(price,period):
                        # 初始化变量（修正角度单位为弧度）
                        a1 = np.exp(-1.414 * np.pi / period)
                        b1 = 2 * a1 * np.cos(1.414 * np.pi / period)  # 将180改为np.pi
                        c2 = b1
                        c3 = -a1 ** 2
                        c1 = (1 + c2 - c3) / 4
                        
                        # 准备输出序列
                        us = np.zeros(len(price))
                        us_new = np.zeros(len(price))
                        trend = [None]*(len(price))
                        ma_close = np.zeros(len(price))
                        
                        # 前4个点用原始价格初始化
                        for i in range(len(price)):
                            if i < 4:
                                us[i] = price.iloc[i]
                            else:
                                # 应用递归公式
                                us[i] = (1 - c1) * price.iloc[i] + (2 * c1 - c2) * price.iloc[i-1] \
                                        - (c1 + c3) * price.iloc[i-2] + c2 * us[i-1] + c3 * us[i-2]
                                
                            us_new = np.around(us, decimals=2)
                            ma_close = price.rolling(window=4*period).mean()#5*

                            # if us_new[i]>price[i] and ma_close[i]>price[i]:
                            #     trend[i] = '空头趋势'
                            # elif us_new[i]<price[i] and ma_close[i]<price[i]:
                            #     trend[i] = '多头趋势'
                            # else:
                            #     trend[i] = '无趋势'

                            if us_new[i] < ma_close.iloc[i]:
                                trend[i] = '空头趋势'
                            elif us_new[i] > ma_close.iloc[i]:
                                trend[i] = '多头趋势'
                            else:
                                trend[i] = '无趋势'

                        
                        return us_new,trend

                    trade_df['终极平滑值'],trade_df['趋势方向'] = ultimate_smoother(trade_df['close'],time_period)#,df['ma_close']
                    
                    trade_df['datetime'] = trade_df['datetime'].dt.strftime("%Y-%m-%d %H:%M:%S")


                    def safe_literal_eval(x):
                        """带异常处理的安全转换"""
                        try:
                            return ast.literal_eval(x)
                        except ValueError:
                            return []  # 返回空列表作为占位符

                    def add_poc_column(df):
                        # 安全转换列数据
                        df['price'] = df['price'].apply(safe_literal_eval)
                        df['Ask'] = df['Ask'].apply(lambda x: list(map(int, safe_literal_eval(x))))
                        df['Bid'] = df['Bid'].apply(lambda x: list(map(int, safe_literal_eval(x))))
                        
                        # 定义处理函数（带数据验证）
                        def find_poc(row):
                            # 验证三个列表长度一致且非空
                            if not (len(row['price']) == len(row['Ask']) == len(row['Bid']) > 0):
                                return '缺值'  # 返回空值标记异常数据
                            
                            sums = [a + b for a, b in zip(row['Ask'], row['Bid'])]
                            try:
                                max_index = sums.index(max(sums))
                                return row['price'][max_index]
                            except ValueError:
                                return '缺值'  # 处理空求和列表情况
                        
                        # 应用处理函数
                        df['POC'] = df.apply(find_poc, axis=1)
                        
                        # 可选：统计异常数据
                        error_count = df['POC'].isnull().sum()
                        if error_count > 0:
                            print(f"警告：发现 {error_count} 行异常数据（已标记为NaN）")
                        
                        return df['POC']

                    trade_df['POC'] = add_poc_column(trade_df)

                    def finall_trend(delta_sum,trend):
                        f_trend = [None]*(len(delta_sum))
                        # delta_sum = delta_sum.astype(float)
                        for i in range(len(delta_sum)):
                            if (delta_sum[i] == '缺值') or (trend[i] == '缺值'):
                                f_trend[i] = '方向不明'
                                # return f_trend
                            else:
                                if delta_sum[i] > 0 and (trend[i] == '多头趋势'):
                                    f_trend[i] = '强多头'
                                elif delta_sum[i] < 0 and (trend[i] == '空头趋势'):
                                    f_trend[i] = '强空头'
                                else:
                                    f_trend[i] = '方向不明'
                        return f_trend   

                    trade_df['最终趋势'] = finall_trend(trade_df['delta累计'],trade_df['趋势方向'])

                    # table_text = f"品种：{trade_df['symbol'].iloc[-1]}, 时间：{trade_df['datetime'].iloc[-1]},close:{trade_df['close'].iloc[-1]},open:{trade_df['open'].iloc[-1]},high:{trade_df['high'].iloc[-1]},low:{trade_df['low'].iloc[-1]},delta:{trade_df['delta'].iloc[-1]}, delta累计:{trade_df['delta累计'].iloc[-1]}, dj:{trade_df['dj'].iloc[-1]},POC:{trade_df['POC'].iloc[-1]},  终极平滑值:{trade_df['终极平滑值'].iloc[-1]}, 趋势方向:{trade_df['趋势方向'].iloc[-1]},最终趋势:{trade_df['最终趋势'].iloc[-1]}"
                    # if data["InstrumentID"]:
                    #     option_buy_symbol,option_buy_price= get_otm_option(data["InstrumentID"], 'C')
                    #     option_sell_symbol,option_sell_price = get_otm_option(data["InstrumentID"], 'P')
                    #     print("买入平值期权为:", option_buy_symbol, "，价格为:", option_buy_price)
                    #     print("卖出平值期权为:", option_sell_symbol, "，价格为:", option_sell_price)

                    def get_otm_option(future_symbol, trade_type):
                        def get_option_symbol(future_symbol):
                            # 创建一个字典,将期货值futuresymbol映射到对应的option symbol .
                            option_dict = {
                                "IH": "上证50股指期权",
                                "IF": "沪深300股指期权",
                                "IC": "中证500股指期权",
                                "IM": "中证1000股指期权"
                            }
                            # 解析 future_symbol 获取期货代码和到期月份
                            m = re.match(r'([A-Za-z]+)(\d+)', future_symbol)
                            if not m:
                                raise ValueError(f"future_symbol 格式不正确: {future_symbol}")
                            future_code, future_end_month = m.groups()
                            option_symbol = option_dict.get(future_code)
                            return option_symbol, future_end_month

                        try:
                            option_symbol, future_end_month = get_option_symbol(future_symbol)
                            option_finance_board_df = ak.option_finance_board(symbol=option_symbol, end_month=future_end_month)

                            len(option_finance_board_df)

                            half = len(option_finance_board_df) // 2

                            first_half_df = option_finance_board_df.iloc[:half]
                            first_half_df.columns = [f"{col}_C" for col in first_half_df.columns]
                            first = first_half_df.reset_index(drop=True)

                            second_half_df = option_finance_board_df.iloc[half:]
                            second_half_df.columns = [f"{col}_P" for col in second_half_df.columns]
                            second = second_half_df.reset_index(drop=True)
                            
                            df = pd.concat([first, second], axis=1)
                            df['lastprice_(C-P)'] = df['lastprice_C'] - df['lastprice_P']

                            idx = df['lastprice_(C-P)'].abs().idxmin()
                            row = df.loc[idx, ['instrument_C', 'instrument_P', 'lastprice_C','lastprice_P','lastprice_(C-P)']]

                            # print(f"index={idx}, instrument_C={row['instrument_C']}, instrument_P={row['instrument_P']}, lastprice_(C-P)={row['lastprice_(C-P)']}")

                            # return df.loc[idx, ['instrument_C', 'instrument_P']]
                            if trade_type == 'C':
                                return row['instrument_C'], float(row['lastprice_C'])
                            elif trade_type == 'P':
                                return row['instrument_P'], float(row['lastprice_P'])
                            else:
                                raise ValueError(f"未知的 trade_type: {trade_type}")
                        except Exception as e:
                            print(f"get_otm_option error for {future_symbol}: {e}")
                            return None

                    def get_otm_pirce(future_symbol, trade_option_symbol):
                        def get_option_symbol(future_symbol):
                            # 创建一个字典,将期货值futuresymbol映射到对应的option symbol .
                            option_dict = {
                                "IH": "上证50股指期权",
                                "IF": "沪深300股指期权",
                                "IC": "中证500股指期权",
                                "IM": "中证1000股指期权"
                            }
                            # 解析 future_symbol 获取期货代码和到期月份
                            m = re.match(r'([A-Za-z]+)(\d+)', future_symbol)
                            if not m:
                                raise ValueError(f"future_symbol 格式不正确: {future_symbol}")
                            future_code, future_end_month = m.groups()
                            option_symbol = option_dict.get(future_code)
                            return option_symbol, future_end_month

                        try:
                            option_symbol, future_end_month = get_option_symbol(future_symbol)
                            option_finance_board_df = ak.option_finance_board(symbol=option_symbol, end_month=future_end_month)
                            
                            mask = option_finance_board_df['instrument'] == trade_option_symbol
                            if mask.any():
                                lastprice_value = option_finance_board_df.loc[mask, 'lastprice'].iloc[0]
                                return  lastprice_value
                                # print(lastprice_value)
                            else:
                                print("未找到对应的行")
                        except Exception as e:
                            print(f"get_otm_option error for {future_symbol}: {e}")
                            return None

                    print("trade_df['symbol'].iloc[-1]}:", trade_df['symbol'].iloc[-1])
                    print("trade_df['symbol'].iloc[-1]}的类型:", type(trade_df['symbol'].iloc[-1]))


                    # 开多、空前置条件
                    开多条件 = (trade_df['趋势方向'].iloc[-1] == '多头趋势')
                    开空条件 = (trade_df['趋势方向'].iloc[-1] == '空头趋势')
                    
                    if len(trade_df) >= 4*time_period:
                        #开多
                        开多1 = (trade_df['dj'].iloc[-1] >= 2) #max(0.8 * max(trade_df['dj'].iloc[-4*time_period-1:-1]), 10))
                        # print("开多1:",开多1)
                        开多2 = (trade_df['delta'].iloc[-1] >= 10)# max(0.8 * max(trade_df['delta'].iloc[-4*time_period-1:-1]), 350))
                        开多3 = (trade_df['delta累计'].iloc[-2] < 0 and trade_df['delta累计'].iloc[-1] > 0)
                        
                        # 开空
                        开空1 = (trade_df['dj'].iloc[-1] <= -2) #min(0.8 * min(trade_df['dj'].iloc[-4*time_period-1:-1]), -10))
                        开空2 = (trade_df['delta'].iloc[-1] <= -10) #min(0.8 * min(trade_df['delta'].iloc[-4*time_period-1:-1]),-350))
                        # print("开空2:",开空2)
                        开空3 = (trade_df['delta累计'].iloc[-2] > 0 and trade_df['delta累计'].iloc[-1] < 0)

                        # 开多组合 = (开多条件 and (开多1 or 开多2 or 开多3))
                        # 开空组合 = (开空条件 and (开空1 or 开空2 or 开空3))

                        # 平多组合 = (开空条件 or 开空1 or 开空2 or 开空3)
                        # 平空组合 = (开多条件 or 开多1 or 开多2 or 开多3)

                        开多组合 = 开多1 and 开多2
                        开空组合 = 开空1 and 开空2

                        平多组合 = 开空1 or 开空2
                        平空组合 = 开多1 or 开多2

                        option_buy_symbol = ''
                        option_buy_price = 0 
                        option_sell_symbol = ''
                        option_sell_price = 0

                        # 平空
                        if param.pos < 0 and 平空组合:
                            # close_sell_price = get_otm_pirce(data["InstrumentID"].decode(), option_sell_symbol)
                            print(
                                "平空：    ",
                                "ExchangeID: ",
                                data["ExchangeID"],
                                "InstrumentID",
                                option_sell_symbol,
                                "AskPrice1",
                                800,#close_sell_price + param.py,
                            )
                            # 平空
                            self.insert_order(
                                data["ExchangeID"],
                                option_sell_symbol,
                                800,#close_sell_price + param.py,
                                param.Lots,
                                b"0",
                                b"1",
                            )
                            self.insert_order(
                                data["ExchangeID"],
                                option_sell_symbol,
                                800,#close_sell_price + 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:{option_sell_symbol}, datetime:{trade_df['datetime'].iloc[-1]}, C_S_T_Price:{data['AskPrice1']  + param.py}, T_Lots:{param.Lots}"
                            send_mail(text)

                        # 开多
                        if param.pos == 0 and 开多组合:
                            print("trade_df_last:",trade_df['symbol'].iloc[-1])
                            option_buy_symbol,option_buy_price= get_otm_option(data["InstrumentID"].decode(), 'C')
                            print("买入看涨平值期权为:", option_buy_symbol, "，价格为:", option_buy_price)
                            print(
                                "开多：    ",
                                "ExchangeID: ",
                                data["ExchangeID"],
                                "InstrumentID",
                                option_buy_symbol,
                                "AskPrice1",
                                option_buy_price + param.py,
                            )
                            # 开多
                            self.insert_order(
                                data["ExchangeID"],
                                option_buy_symbol,
                                option_buy_price + param.py,
                                param.Lots,
                                b"0",
                                b"0",
                            )
                            print(
                                "datetime+sig: ",
                                trade_df["datetime"].iloc[-1],
                                "多头开仓",
                                "开仓价格：",
                                option_buy_price + 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:{option_buy_symbol}, datetime:{trade_df['datetime'].iloc[-1]}, O_L_T_Price:{option_buy_price + param.py}, T_Lots:{param.Lots}"
                            send_mail(text)

                        # 平多
                        if param.pos > 0 and 平多组合:
                            print('option_buy_symbol',option_buy_symbol)
                            # close_buy_price = get_otm_pirce(data["InstrumentID"].decode(), option_buy_symbol)
                            # print('close_buy_price:',close_buy_price)
                            print(
                                "平多：   ",
                                "ExchangeID: ",
                                data["ExchangeID"],
                                "InstrumentID",
                                option_buy_symbol,
                                "BidPrice1",
                                1,#close_buy_price - param.py,
                            )
                            # 平多
                            self.insert_order(
                                data["ExchangeID"],
                                option_buy_symbol,
                                1,#close_buy_price - param.py,
                                param.Lots,
                                b"1",
                                b"1",
                            )
                            self.insert_order(
                                data["ExchangeID"],
                                option_buy_symbol,
                                1,#close_buy_price - 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:{option_buy_symbol}, datetime:{trade_df['datetime'].iloc[-1]}, C_L_T_Price:{data['BidPrice1'] - param.py}, T_Lots:{param.Lots}"
                            send_mail(text)

                        # 开空
                        if param.pos == 0 and 开空组合:
                            option_sell_symbol,option_sell_price = get_otm_option(data["InstrumentID"].decode(), 'P')
                            print("买入看跌平值期权为:", option_sell_symbol, "，价格为:", option_sell_price)
                            print(
                                "开空：   ",
                                "ExchangeID: ",
                                data["ExchangeID"],
                                "InstrumentID",
                                option_sell_symbol,
                                "BidPrice1",
                                data["BidPrice1"],
                            )
                            # 开空
                            self.insert_order(
                                data["ExchangeID"],
                                option_sell_symbol,
                                option_sell_price - param.py,
                                param.Lots,
                                b"1",
                                b"0",
                            )
                            print(
                                "datetime+sig: ",
                                trade_df["datetime"].iloc[-1],
                                "空头开仓",
                                "开仓价格：",
                                option_sell_price - 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:{option_sell_symbol}, datetime:{trade_df['datetime'].iloc[-1]}, O_S_T_Price:{option_sell_price - param.py}, T_Lots:{param.Lots}"
                            send_mail(text)

                    print(trade_df)
                    symbol_label = trade_df["symbol"].iloc[-1]
                    trade_df.to_csv(f"trade_df_{symbol_label}.csv", index=False, encoding='utf-8')
                    # with open('trade_df.txt', 'w', encoding='utf-8') as file:
                    #     print(trade_df, file=file)
                    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["TF2509"] = ParamObj(
    #     symbol="TF2509",
    #     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",
    # )

    ## 交易所有品种，自动设置合约代码,
    # 交易指定品种时，symbols = ['IM','IC','ag']
    # symbols = ['IM','IC','ag']
    symbols = contacts_df['品种代码'].tolist()
    # for i, symbol in enumerate(symbols, start=1):
        # globals()[f'sb_{i}'] = get_main_contact_on_time(symbol, contacts_df)
        # symbol = globals()[f'sb_{i}']
        # # print("最终使用的主连代码：",symbol)
        # param_dict[str(symbol)] = ParamObj(symbol=symbol.encode('ascii'),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,周期="5min")
    for i, symbol in enumerate(symbols, start=1):
        if symbol in ['wr', 'RS' , 'bb', 'WH', 'fb', 'rr', 'PL']:
            continue
        elif symbol in ['IM',]: #symbol in ['IH', 'IF', 'IC', 'IM', 'au', 'sc']
            globals()[f'sb_{i}'] = get_main_contact_on_time(symbol, contacts_df)
            symbol = globals()[f'sb_{i}']
            param_dict[str(symbol)] = ParamObj(symbol=symbol.encode('ascii'),Lots=1,py=5,trailing_stop_percent=0.02,fixed_stop_loss_percent=0.04,dj_X=8,delta=500,sum_delta=800,失衡=3,堆积=3,周期="2min")
    

    ## 交易多个指定品种，自动设置合约代码,手动设置其他参数
    # param_dict[symbol] = ParamObj(symbol=get_main_contact_on_time('IM', contacts_df),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,周期="1min")
    # param_dict[symbol] = ParamObj(symbol=get_main_contact_on_time('IC', contacts_df),Lots=1,py=5,trailing_stop_percent=0.02,fixed_stop_loss_percent=0.04,dj_X=8,delta=500,sum_delta=800,失衡=3,堆积=3,周期="1min")
    # print(param_dict.keys())
    
    # 用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="227508",  # simnow账户，注意是登录账户的ID，SIMNOW个人首页查看
        password="Zj1234!@#%",  # simnow密码
        server_name="电信1",  # 电信1、电信2、移动、TEST、N视界
        subscribe_list=list(param_dict.keys()),  # 合约列表
    )
    # future_account = get_simulate_account(
    #     investor_id="00033556",  # simnow账户，注意是登录账户的ID，SIMNOW个人首页查看
    #     password="27138169",  # simnow密码
    #     server_name="N视界",  # 电信1、电信2、移动、TEST、N视界
    #     subscribe_list=list(param_dict.keys()),  # 合约列表
    # )
    # 实盘用这个，不要忘记屏蔽上方simnow的future_account字典
    # future_account = FutureAccount(
    #     broker_id='',  # 期货公司BrokerID
    #     server_dict={'TDServer': "121.37.80.177:20002", 'MDServer': '121.37.80.177:20004'},  # TDServer为交易服务器，MDServer为行情服务器。服务器地址格式为"ip:port。"
    #     reserve_server_dict={},  # 备用服务器地址
    #     investor_id='1114',  # 账户
    #     password='123456',  # 密码
    #     app_id='',  # 认证使用AppID
    #     auth_code='',  # 认证使用授权码
    #     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()
