using BaseOUDAL;
using Confluent.Kafka;
using DocumentFormat.OpenXml.Spreadsheet;
using DocumentFormat.OpenXml.VariantTypes;
using Microsoft.Office.Interop.Excel;
using Newtonsoft.Json;
using NPOI.SS.Formula.Functions;
using Qdp.Pricing.Library.Base.Utilities;
using System;
using System.Collections.Generic;
using System.Drawing.Drawing2D;
using System.Linq;
using System.Linq.Dynamic.Core;
using System.Linq.Expressions;
using System.Text;
using System.Threading.Tasks;
using YLErp.Abstract;
using YLErp.DBModels;
using YLErp.DBModels.Enums;
using YLErp.Helpers;
using YLErp.Model;
using YLErp.Model.HengTaiModel;
using YLErp.Models;
using YLErp.Modules.AppModule;
using YLErp.Modules.ClientModule;
using YLErp.Modules.FinancialModule;
using YLErp.QdpModule;
using static YLErp.DBModels.ConsTrade;

namespace YLErp.Modules.SwapModule
{
    public class SwapPushService : YLBaseService
    {
        // 互换交易kafkaTopic
        private string reqInterestRateSwapInsert = string.Empty;
        // 资产kafkaTopic
        private string reqAssetSwapInsert = string.Empty;
        // 预付金kafkaTopic
        private string reqMarginInsert = string.Empty;

        //平仓kafkaTopic
        private string reqAcctSwapTerminate = string.Empty;

        //交易对手方kafkaTopic
        private string reqCounterPartyInfoInsert = string.Empty;

        //交易对手方资金账户
        private string reqCounterPartyCashAccoutInsert = string.Empty;

        List<int> liveLogStates = new List<int>() { (int)SwapPushDataStateEnum.待推送 };
        private IKafkaProduce kafkaProduceHelper;
        public SwapPushService(OptUserInfo optUser) : base(optUser)
        {
            reqInterestRateSwapInsert = Environment.GetEnvironmentVariable("KafkaConfig_ReqInterestRateSwapInsertTopic");
            reqAssetSwapInsert = Environment.GetEnvironmentVariable("KafkaConfig_ReqAssetSwapInsertTopic");
            reqMarginInsert = Environment.GetEnvironmentVariable("KafkaConfig_ReqMarginInsertTopic");
            reqAcctSwapTerminate = Environment.GetEnvironmentVariable("KafkaConfig_ReqAcctSwapTerminateTopic");
            reqCounterPartyInfoInsert = Environment.GetEnvironmentVariable("KafkaConfig_ReqCounterPartyInfoInsertTopic");
            reqCounterPartyCashAccoutInsert = Environment.GetEnvironmentVariable("KafkaConfig_ReqCounterPartyCashAccoutInsertTopic");
        }
        public void SetKafKaProduce(IKafkaProduce kafkaProduce)
        {
            kafkaProduceHelper = kafkaProduce;
        }
        /// <summary>
        /// 推送交易
        /// </summary>
        public void PushTrade()
        {
            PushData(kafkaProduceHelper, reqInterestRateSwapInsert, SwapPushDataEnum.收益互换交易新增, "2042", PrepareTradeReq);
        }
        /// <summary>
        /// 推送交易
        /// </summary>
        public void PushTradeUpdate()
        {
            PushData(kafkaProduceHelper, reqInterestRateSwapInsert, SwapPushDataEnum.收益互换交易修改, "3005", PrepareTradeReq);
            PushPosition(kafkaProduceHelper, SwapPushDataEnum.互换资产交易修改, "3005");
            PushMargin(kafkaProduceHelper, SwapPushDataEnum.预付金交易修改, "3005");
        }
        public void RePushLog(PushLog pushLog)
        {
            List<PushLog> pushLogs = new List<PushLog>() { pushLog };
            var pushDataType = pushLog.data_type;
            if (pushDataType == (int)SwapPushDataEnum.收益互换交易新增)
            {
                PushData(kafkaProduceHelper, reqInterestRateSwapInsert, pushLogs, "2042", PrepareTradeReq);
            }
            else if (pushDataType == (int)SwapPushDataEnum.收益互换交易修改)
            {
                PushData(kafkaProduceHelper, reqInterestRateSwapInsert, pushLogs, "3005", PrepareTradeReq);
            }
            else if (pushDataType == (int)SwapPushDataEnum.互换资产交易新增)
            {
                PushPositionData(kafkaProduceHelper, pushLogs, "2041");
            }
            else if (pushDataType == (int)SwapPushDataEnum.互换资产交易修改)
            {
                PushPositionData(kafkaProduceHelper, pushLogs, "3005");
            }
            else if (pushDataType == (int)SwapPushDataEnum.互换资产平仓)
            {
                PushUnwindPositionData(kafkaProduceHelper, pushLogs, true);
            }
            else if (pushDataType == (int)SwapPushDataEnum.预付金交易新增)
            {
                PushMarginData(kafkaProduceHelper, pushLogs, "2043");
            }
            else if (pushDataType == (int)SwapPushDataEnum.预付金交易修改)
            {
                PushMarginData(kafkaProduceHelper, pushLogs, "3005");
            }
            else if (pushDataType == (int)SwapPushDataEnum.互换预付金平仓)
            {
                PushUnwindMarginData(kafkaProduceHelper, pushLogs, true);
            }
            else if (pushDataType == (int)SwapPushDataEnum.客户创建 || pushDataType == (int)SwapPushDataEnum.客户修改)
            {
                PushClientData(kafkaProduceHelper, pushLogs);
            }
            else if (pushDataType == (int)SwapPushDataEnum.客户资金账户新增 || pushDataType == (int)SwapPushDataEnum.客户资金账户修改 || pushDataType == (int)SwapPushDataEnum.客户资金账户删除)
            {
                PushClientBannkData(kafkaProduceHelper, pushLogs);
            }
            else if (pushDataType == (int)SwapPushDataEnum.合约终止)
            {
                PushUnwindData(kafkaProduceHelper, pushLogs, "T");
            }
            else if (pushDataType == (int)SwapPushDataEnum.合约撤单)
            {
                PushRepealUnwind(kafkaProduceHelper, pushLogs);
            }
            //else if (pushDataType == (int)SwapPushDataEnum.客户入金新增)
            //{
            //    PushRepealCash(kafkaProduceHelper, pushLogs);
            //}
            //else if (pushDataType == (int)SwapPushDataEnum.客户出金新增)
            //{
            //    PushRepealCash(kafkaProduceHelper, pushLogs);
            //}
        }
        /// <summary>
        /// 交易对手方推送
        /// </summary>
        /// <param name="kafkaProduceHelper"></param>
        public void PushClient()
        {
            var pushService = new PushLogService(UserInfo);
            List<int> dataTypes = new List<int>() { (int)SwapPushDataEnum.客户创建, (int)SwapPushDataEnum.客户修改 };
            var pushLogs = pushService.GetPushLogs(dataTypes, liveLogStates);
            PushClientData(kafkaProduceHelper, pushLogs);
        }
        /// <summary>
        /// 交易对手方银行卡信息推送
        /// </summary>
        public void PushClientBank()
        {
            var pushService = new PushLogService(UserInfo);
            List<int> dataTypes = new List<int>() { (int)SwapPushDataEnum.客户资金账户修改, (int)SwapPushDataEnum.客户资金账户删除 };
            var pushLogs = pushService.GetPushLogs(dataTypes, liveLogStates);
            PushClientBannkData(kafkaProduceHelper, pushLogs);
        }
        /// <summary>
        /// 推送客户资金记录
        /// </summary>
        public void PushClientCash()
        {
            //var pushService = new PushLogService(UserInfo);
            //List<int> dataTypes = new List<int>() { (int)SwapPushDataEnum.客户入金新增, (int)SwapPushDataEnum.客户出金新增 };
            //var pushLogs = pushService.GetPushLogs(dataTypes, liveLogStates);
            //PushRepealCash(kafkaProduceHelper, pushLogs);
        }
        /// <summary>
        /// 消费推送结果
        /// </summary>
        public void ConsumerTradeResp(KafkaConsumerHelper _kafkaConsumer)
        {
            _kafkaConsumer.Subscribe(msg =>
            {
                if (!string.IsNullOrEmpty(msg))
                {
                    var result = JsonHelper.Deserialize<ResponseBase>(msg);
                    var logid = Convert.ToInt64(result.HEADER.SERIAL_NO);
                    int resultState = (int)SwapPushDataStateEnum.推送失败;
                    if (result.HEADER.ERRCODE == "0" && result.RECORD.RESULT == "0")
                    {
                        resultState = (int)SwapPushDataStateEnum.推送成功;
                    }
                    var pushService = new PushLogService(UserInfo);
                    var pushLog = pushService.UpdateLogResp(logid, msg, resultState);
                    if (resultState == (int)SwapPushDataStateEnum.推送成功 && pushLog != null)
                    {
                        if (pushLog.data_type == (int)SwapPushDataEnum.收益互换交易新增)
                        {
                            var posiNumbers = DbContext.swap_position.Where(x => x.SwapTradeId == pushLog.data_id && x.IsInitial && !x.Invalid).Select(s => s.PosiNumber).ToList();
                            Expression<Func<PushLog, bool>> expression = n => n.data_type == (int)SwapPushDataEnum.互换资产交易新增 && posiNumbers.Contains(n.data_number);
                            Expression<Func<PushLog, bool>> expression2 = n => n.data_type == (int)SwapPushDataEnum.预付金交易新增 && posiNumbers.Contains(n.data_number);
                            var posiPushLogs = pushService.GetPushLogs(expression);
                            var marginPushLogs = pushService.GetPushLogs(expression2);
                            PushPositionData(kafkaProduceHelper, posiPushLogs, "2041");
                            PushMarginData(kafkaProduceHelper, marginPushLogs, "2043");
                        }
                    }
                }
            });
        }
        /// <summary>
        /// 消费交易对手方推送结果
        /// </summary>
        /// <param name="_kafkaConsumer"></param>
        public void ConsumerClientResp(KafkaConsumerHelper _kafkaConsumer)
        {
            _kafkaConsumer.Subscribe(msg =>
            {
                if (!string.IsNullOrEmpty(msg))
                {
                    var result = JsonHelper.Deserialize<ClientPushResp>(msg);
                    var logid = Convert.ToInt64(result.HEADER.SERIAL_NO);
                    int resultState = (int)SwapPushDataStateEnum.推送失败;
                    string irPartyId = string.Empty;
                    if (result.HEADER.ERRCODE == "0")
                    {
                        resultState = (int)SwapPushDataStateEnum.推送成功;
                        if (result.RECORD.PARSEINFO.Count > 0)
                        {
                            if (result.RECORD.PARSEINFO.Any(a => a.RESULT != "1"))
                            {
                                resultState = (int)SwapPushDataStateEnum.推送失败;
                            }
                            else
                            {
                                irPartyId = result.RECORD.PARSEINFO[0].XIR_PARTY_ID;
                            }
                        }
                    }
                    var pushService = new PushLogService(UserInfo);
                    var pushLog = pushService.UpdateLogResp(logid, msg, resultState);
                    if (resultState == (int)SwapPushDataStateEnum.推送成功)
                    {
                        new ClientSaveService(UserInfo).UpdateClientOutNumber(pushLog.data_number, irPartyId);
                        if (pushLog.data_type == (int)SwapPushDataEnum.客户创建)
                        {
                            Expression<Func<PushLog, bool>> expression = n => n.data_type == (int)SwapPushDataEnum.客户资金账户新增 && n.data_number == pushLog.data_number;
                            var pushLogs = pushService.GetPushLogs(expression);
                            PushClientData(kafkaProduceHelper, pushLogs);
                        }
                    }
                }
            });
        }
        /// <summary>
        /// 消费交易对手方推送结果
        /// </summary>
        /// <param name="_kafkaConsumer"></param>
        public void ConsumerClientBankResp(KafkaConsumerHelper _kafkaConsumer)
        {
            _kafkaConsumer.Subscribe(msg =>
            {
                if (!string.IsNullOrEmpty(msg))
                {
                    var result = JsonHelper.Deserialize<ClientPushResp>(msg);
                    var logid = Convert.ToInt64(result.HEADER.SERIAL_NO);
                    int resultState = (int)SwapPushDataStateEnum.推送失败;
                    if (result.HEADER.ERRCODE == "0")
                    {
                        resultState = (int)SwapPushDataStateEnum.推送成功;
                        if (result.RECORD.PARSEINFO.Count > 0)
                        {
                            if (result.RECORD.PARSEINFO.Any(a => a.RESULT != "1"))
                            {
                                resultState = (int)SwapPushDataStateEnum.推送失败;
                            }
                        }
                    }
                    new PushLogService(UserInfo).UpdateLogResp(logid, msg, resultState);
                }
            });
        }
        /// <summary>
        /// 推送平仓
        /// </summary>
        /// <param name="kafkaProduceHelper"></param>
        public void PushUnwind()
        {
            PushUnwindPosi(kafkaProduceHelper, SwapPushDataEnum.互换资产平仓, PushUnwindPositionData);
            PushUnwindPosi(kafkaProduceHelper, SwapPushDataEnum.互换预付金平仓, PushUnwindMarginData);
            PushUnwindData(kafkaProduceHelper, SwapPushDataEnum.合约终止, "T");
        }
        /// <summary>
        /// 平仓或回退
        /// </summary>
        /// <param name="kafkaProduceHelper"></param>
        /// <param name="swapPushDataEnum"></param>
        /// <param name="operate"></param>
        private void PushUnwindData(IKafkaProduce kafkaProduceHelper, SwapPushDataEnum swapPushDataEnum, string operate)
        {
            var pushService = new PushLogService(UserInfo);
            List<int> dataTypes = new List<int>() { (int)swapPushDataEnum };
            var pushLogs = pushService.GetPushLogs(dataTypes, liveLogStates);
            PushUnwindData(kafkaProduceHelper, pushLogs, operate);
        }
        /// <summary>
        /// 合约平仓推送
        /// </summary>
        /// <param name="kafkaProduceHelper"></param>
        /// <param name="pushLogs"></param>
        /// <param name="operate"></param>
        private void PushUnwindData(IKafkaProduce kafkaProduceHelper, List<PushLog> pushLogs, string operate)
        {
            var pushService = new PushLogService(UserInfo);
            var delPushIds = new List<long>();
            foreach (var pushLog in pushLogs)
            {
                var swapEvent = DbContext.swap_event.FirstOrDefault(x => x.id == pushLog.data_id);
                if (swapEvent == null)
                {
                    pushService.DeleteLog(pushLog.id);
                    continue;
                }
                var trade = DbContext.trade.FirstOrDefault(x => x.id == swapEvent.SwapTradeId);
                if (trade == null)
                {
                    pushService.DeleteLog(pushLog.id);
                    continue;
                }
                if (trade.ValidState == ConsGlobal.InValid)
                {
                    var number = trade.TradeNumber.Replace("XX","");
                    var tradePushs = pushService.GetPushLogs(x => x.data_number.StartsWith(number));
                    if (!tradePushs.Any(x => x.data_state == (int)SwapPushDataStateEnum.推送成功))//已经删除的交易没有推送成功的情况下全部删除记录
                    {
                        delPushIds.AddRange(tradePushs.Select(s => s.id));
                        pushService.DeleteLogs(delPushIds);
                        continue;
                    }
                }
                var req = new RequestBase<SwapUnwindReq>();
                swapEvent.unwindData = JsonConvert.DeserializeObject<UnwindData>(swapEvent.EventData);
                swapEvent.unwindData.FlowEvents = DbContext.swap_flow_event.Where(x => x.EventId == swapEvent.id).ToList();
                req.RECORD = PrepareUnwindData(swapEvent);
                if (req.RECORD == null)
                {
                    pushService.DeleteLog(pushLog.id);
                    continue;
                }
                req.EXT_NO = trade.TradeNumber.Replace("XX", "");
                req.SERIAL_NO = pushLog.id.ToString();
                req.OPERATE = operate;
                var data_req = JsonHelper.Serialize(req);
                pushService.UpdateLogReq(pushLog.id, data_req);
                kafkaProduceHelper.Produce(reqAcctSwapTerminate, data_req);
            }
        }
        /// <summary>
        /// 资产平仓推送
        /// </summary>
        /// <param name="kafkaProduceHelper"></param>
        /// <param name="swapPushDataEnum"></param>
        private void PushUnwindPosi(IKafkaProduce kafkaProduceHelper, SwapPushDataEnum swapPushDataEnum, Action<IKafkaProduce, List<PushLog>, bool> actionMethod)
        {
            var pushService = new PushLogService(UserInfo);
            List<int> dataTypes = new List<int>() { (int)swapPushDataEnum };
            var pushLogs = pushService.GetPushLogs(dataTypes, liveLogStates);
            actionMethod(kafkaProduceHelper, pushLogs, true);
        }
        /// <summary>
        /// 资产平仓推送
        /// </summary>
        /// <param name="kafkaProduceHelper"></param>
        /// <param name="pushLogs"></param>
        private void PushUnwindPositionData(IKafkaProduce kafkaProduceHelper, List<PushLog> pushLogs, bool unwind)
        {
            var pushService = new PushLogService(UserInfo);
            var delPushIds = new List<long>();
            foreach (var pushLog in pushLogs)
            {
                if (delPushIds.Contains(pushLog.id))
                {
                    continue;
                }
                var flowEvent = DbContext.swap_flow_event.FirstOrDefault(x => x.id == pushLog.data_id);
                if (flowEvent == null)
                {
                    pushService.DeleteLog(pushLog.id);
                    continue;
                }
                var trade = DbContext.trade.FirstOrDefault(x => x.id == flowEvent.SwapTradeId);
                if (trade == null)
                {
                    pushService.DeleteLog(pushLog.id);
                    continue;
                }
                if (trade.ValidState == ConsGlobal.InValid)
                {
                    var number = trade.TradeNumber.Replace("XX", "");
                    var tradePushs = pushService.GetPushLogs(x => x.data_number.StartsWith(number));
                    if (!tradePushs.Any(x => x.data_state == (int)SwapPushDataStateEnum.推送成功))//已经删除的交易没有推送成功的情况下全部删除记录
                    {
                        delPushIds.AddRange(tradePushs.Select(s => s.id));
                        pushService.DeleteLogs(delPushIds);
                        continue;
                    }
                }
                var client = DataCacheProvider.GetClientDataSource().GetData(trade.ClientId);
                var req = new RequestBase<SwapPosiReq>();
                req.RECORD = PrepareTradePosiUnwindReq(trade, flowEvent, pushLog.data_number);
                if (req.RECORD == null)
                {
                    pushService.DeleteLog(pushLog.id);
                    continue;
                }
                req.EXT_NO = pushLog.data_id.ToString();
                req.SERIAL_NO = pushLog.id.ToString();
                req.ACTION = "2041";
                var data_req = JsonHelper.Serialize(req);
                pushService.UpdateLogReq(pushLog.id, data_req);
                kafkaProduceHelper.Produce(reqAssetSwapInsert, data_req);
            }
        }

        /// <summary>
        /// 预付金平仓推送
        /// </summary>
        /// <param name="kafkaProduceHelper"></param>
        /// <param name="pushLogs"></param>
        private void PushUnwindMarginData(IKafkaProduce kafkaProduceHelper, List<PushLog> pushLogs, bool unwind)
        {
            var pushService = new PushLogService(UserInfo);
            foreach (var pushLog in pushLogs)
            {
                var flowEvent = DbContext.swap_flow_event.FirstOrDefault(x => x.id == pushLog.data_id);
                if (flowEvent == null)
                {
                    pushService.DeleteLog(pushLog.id);
                    continue;
                }
                var trade = DbContext.trade.FirstOrDefault(x => x.id == flowEvent.SwapTradeId);
                var client = DataCacheProvider.GetClientDataSource().GetData(trade.ClientId);
                var req = new RequestBase<SwapMarginReq>();
                req.RECORD = PrepareTradeMarginUnwindReq(client, trade, flowEvent, unwind, pushLog.data_number);
                if (req.RECORD == null)
                {
                    pushService.DeleteLog(pushLog.id);
                    continue;
                }
                req.EXT_NO = pushLog.data_id.ToString();
                req.SERIAL_NO = pushLog.id.ToString();
                req.ACTION = "2043";
                var data_req = JsonHelper.Serialize(req);
                pushService.UpdateLogReq(pushLog.id, data_req);
                kafkaProduceHelper.Produce(reqMarginInsert, data_req);
            }
        }

        /// <summary>
        /// 推送回退
        /// </summary>
        /// <param name="kafkaProduceHelper"></param>
        public void PushRepealUnwind()
        {
            var pushService = new PushLogService(UserInfo);
            List<int> dataTypes = new List<int>() { (int)SwapPushDataEnum.合约撤单 };
            var pushLogs = pushService.GetPushLogs(dataTypes, liveLogStates);
            PushRepealUnwind(kafkaProduceHelper, pushLogs);
        }
        private void PushRepealUnwind(IKafkaProduce kafkaProduceHelper, List<PushLog> pushLogs)
        {
            var pushService = new PushLogService(UserInfo);
            foreach (var pushLog in pushLogs)
            {
                var trade = new trade();
                if (pushLog.data_number.Contains("-"))
                {
                    var position = DbContext.swap_position.FirstOrDefault(x => x.id == pushLog.data_id);
                    if (position == null)
                    {
                        pushService.DeleteLog(pushLog.id);
                        continue;
                    }
                    var hasInterest = DbContext.swap_position.Any(x => x.SwapTradeId == position.SwapTradeId && x.IsInitial && !x.Invalid && ConsTrade.InterestNotionalModels.Contains(x.InterestMode));
                    PushRepealUnwindData(kafkaProduceHelper, position, hasInterest, pushLog, position.PosiNumber, pushService);
                }
                else
                {
                    trade = DbContext.trade.FirstOrDefault(x => x.id == pushLog.data_id);
                    if (trade == null)
                    {
                        pushService.DeleteLog(pushLog.id);
                        continue;
                    }
                    var positions = DbContext.swap_position.Where(x => x.SwapTradeId == trade.id && x.IsInitial).ToList();
                    var tradeNumber = trade.TradeNumber.Replace("XX", "CW");
                    var hasInterest = positions.Any(x => ConsTrade.InterestNotionalModels.Contains(x.InterestMode));
                    var posi = positions.Where(x => x.PosiDirection > 0).FirstOrDefault();
                    if (posi == null)
                    {
                        pushService.DeleteLog(pushLog.id);
                        continue;
                    }
                    PushRepealUnwindData(kafkaProduceHelper, posi, hasInterest, pushLog, tradeNumber, pushService);
                }

            }
        }
        private void PushRepealCash(IKafkaProduce kafkaProduceHelper, List<PushLog> pushLogs)
        {
            var consumerService = new SwapConsumerService(UserInfo);
            consumerService.SetKafKaProduce(kafkaProduceHelper);
            foreach (var pushLog in pushLogs)
            {
                //consumerService.PushCashToHT((int)pushLog.data_id,pushLog);
            }
        }
        private void PushRepealUnwindData(IKafkaProduce kafkaProduceHelper, swap_position posi, bool hasInterest, PushLog pushLog, string tradeNumber, PushLogService pushService)
        {
            var req = new RequestBase<SwapUnwindReq>();
            req.RECORD = PrepareRepealUnwindData(posi.id.ToString(), posi, hasInterest);
            req.EXT_NO = tradeNumber;
            req.SERIAL_NO = pushLog.id.ToString();
            req.OPERATE = "D";
            var data_req = JsonHelper.Serialize(req);
            pushService.UpdateLogReq(pushLog.id, data_req);
            kafkaProduceHelper.Produce(reqAcctSwapTerminate, data_req);
        }
        /// <summary>
        /// 构建数据
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="kafkaProduceHelper"></param>
        /// <param name="topic"></param>
        /// <param name="swapPushDataEnum"></param>
        /// <param name="funcMethod"></param>
        private void PushData<T>(IKafkaProduce kafkaProduceHelper, string topic, SwapPushDataEnum swapPushDataEnum, string action, Func<SwapTradePushModel, T> funcMethod)
        {
            var pushService = new PushLogService(UserInfo);
            List<int> dataTypes = new List<int>() { (int)swapPushDataEnum };
            var pushLogs = pushService.GetPushLogs(dataTypes, liveLogStates);
            var delPushIds = new List<long>();
            foreach (var pushLog in pushLogs)
            {
                if (delPushIds.Contains(pushLog.id))
                {
                    continue;
                }
                trade trade = DbContext.trade.FirstOrDefault(x => x.id == pushLog.data_id);
                if (trade == null)
                {
                    pushService.DeleteLog(pushLog.id);
                    continue;
                }
                if (trade.ValidState == ConsGlobal.InValid)
                {
                    var number = trade.TradeNumber.Replace("XX", "");
                    var tradePushs = pushService.GetPushLogs(x => x.data_number.StartsWith(number));
                    if (!tradePushs.Any(x => x.data_state == (int)SwapPushDataStateEnum.推送成功))//已经删除的交易没有推送成功的情况下全部删除记录
                    {
                        delPushIds.AddRange(tradePushs.Select(s => s.id));
                        pushService.DeleteLogs(delPushIds);
                        continue;
                    }
                }
                var tradeExtend = DbContext.trade_extend.FirstOrDefault(x => x.TradeId == trade.id);
                List<swap_position> positions = DbContext.swap_position.Where(x => x.SwapTradeId == trade.id && x.IsInitial && !x.Invalid).ToList();
                var client = DataCacheProvider.GetClientDataSource().GetData(trade.ClientId);
                SwapTradePushModel model = new SwapTradePushModel() { trade = trade, positions = positions, tradeExtend = tradeExtend, Client = client };
                var req = new RequestBase<T>();
                req.RECORD = funcMethod(model);
                if (req.RECORD == null)
                {
                    pushService.DeleteLog(pushLog.id);
                    continue;
                }
                req.EXT_NO = model.extNo;
                req.SERIAL_NO = pushLog.id.ToString();
                req.ACTION = action;
                var data_req = JsonHelper.Serialize(req);
                pushService.UpdateLogReq(pushLog.id, data_req);
                kafkaProduceHelper.Produce(topic, data_req);
            }
        }

        private void PushData<T>(IKafkaProduce kafkaProduceHelper, string topic, List<PushLog> pushLogs, string action, Func<SwapTradePushModel, T> funcMethod)
        {
            var pushService = new PushLogService(UserInfo);
            foreach (var pushLog in pushLogs)
            {
                trade trade =  DbContext.trade.FirstOrDefault(x => x.id == pushLog.data_id);
                var tradeExtend = DbContext.trade_extend.FirstOrDefault(x => x.TradeId == trade.id);
                List<swap_position> positions = positions = DbContext.swap_position.Where(x => x.SwapTradeId == trade.id && x.IsInitial && !x.Invalid).ToList();
                var client = DataCacheProvider.GetClientDataSource().GetData(trade.ClientId);
                SwapTradePushModel model = new SwapTradePushModel() { trade = trade, positions = positions, tradeExtend = tradeExtend, Client = client };
                var req = new RequestBase<T>();
                req.RECORD = funcMethod(model);
                if (req.RECORD == null)
                {
                    pushService.DeleteLog(pushLog.id);
                    continue;
                }
                req.EXT_NO = model.extNo;
                req.SERIAL_NO = pushLog.id.ToString();
                req.ACTION = action;
                var data_req = JsonHelper.Serialize(req);
                pushService.UpdateLogReq(pushLog.id, data_req);
                kafkaProduceHelper.Produce(topic, data_req);
            }
        }

        /// <summary>
        /// 预付金推送
        /// </summary>
        /// <param name="kafkaProduceHelper"></param>
        /// <param name="swapPushDataEnum"></param>
        /// <param name="action"></param>
        private void PushMargin(IKafkaProduce kafkaProduceHelper, SwapPushDataEnum swapPushDataEnum, string action)
        {
            var pushService = new PushLogService(UserInfo);
            List<int> dataTypes = new List<int>() { (int)swapPushDataEnum };
            var pushLogs = pushService.GetPushLogs(dataTypes, liveLogStates);
            PushMarginData(kafkaProduceHelper, pushLogs, action);
        }
        /// <summary>
        /// 预付金推送
        /// </summary>
        /// <param name="kafkaProduceHelper"></param>
        /// <param name="pushLogs"></param>
        /// <param name="action"></param>
        private void PushMarginData(IKafkaProduce kafkaProduceHelper, List<PushLog> pushLogs, string action)
        {
            var pushService = new PushLogService(UserInfo);
            foreach (var pushLog in pushLogs)
            {
                var position = DbContext.swap_position.FirstOrDefault(x => x.id == pushLog.data_id);
                if (position == null)
                {
                    pushService.DeleteLog(pushLog.id);
                    continue;
                }
                var trade = DbContext.trade.FirstOrDefault(x => x.id == position.SwapTradeId);
                if (trade == null)
                {
                    pushService.DeleteLog(pushLog.id);
                    continue;
                }
                var positions = new List<swap_position>() { position };
                var client = DataCacheProvider.GetClientDataSource().GetData(trade.ClientId);
                SwapTradePushModel model = new SwapTradePushModel() { trade = trade, positions = positions, Client = client };
                var req = new RequestBase<SwapMarginReq>();
                req.RECORD = PrepareTradeMarginReq(model);
                if (req.RECORD == null)
                {
                    pushService.DeleteLog(pushLog.id);
                    continue;
                }
                req.EXT_NO = position.PosiNumber;
                req.SERIAL_NO = pushLog.id.ToString();
                req.ACTION = action;
                var data_req = JsonHelper.Serialize(req);
                pushService.UpdateLogReq(pushLog.id, data_req);
                kafkaProduceHelper.Produce(reqMarginInsert, data_req);
            }
        }
        /// <summary>
        /// 资产推送
        /// </summary>
        /// <param name="kafkaProduceHelper"></param>
        /// <param name="swapPushDataEnum"></param>
        /// <param name="action"></param>
        private void PushPosition(IKafkaProduce kafkaProduceHelper, SwapPushDataEnum swapPushDataEnum, string action)
        {
            var pushService = new PushLogService(UserInfo);
            List<int> dataTypes = new List<int>() { (int)swapPushDataEnum };
            var pushLogs = pushService.GetPushLogs(dataTypes, liveLogStates);
            PushPositionData(kafkaProduceHelper, pushLogs, action);
        }

        /// <summary>
        /// 资产推送推送
        /// </summary>
        /// <param name="kafkaProduceHelper"></param>
        /// <param name="pushLogs"></param>
        /// <param name="action"></param>
        private void PushPositionData(IKafkaProduce kafkaProduceHelper, List<PushLog> pushLogs, string action)
        {
            var pushService = new PushLogService(UserInfo);
            foreach (var pushLog in pushLogs)
            {
                var position = DbContext.swap_position.FirstOrDefault(x => x.id == pushLog.data_id);
                if (position == null)
                {
                    var openQuery = from se in DbContext.swap_flow_event
                                    join p in DbContext.swap_position on se.PositionId equals p.PositionId
                                    where se.id == pushLog.data_id
                                    select new { position = p, flowEvent = se };
                    var open = openQuery.FirstOrDefault();
                    if (open == null)
                    {
                        pushService.DeleteLog(pushLog.id);
                        continue;
                    }
                    position = open.position;
                    position.PosiQuantity = open.flowEvent.PositionQty ?? 0;
                    position.PosiNotionalValue = position.PosiQuantity * position.ContractSize;
                }
                var trade = DbContext.trade.FirstOrDefault(x => x.id == position.SwapTradeId);
                if (trade == null)
                {
                    pushService.DeleteLog(pushLog.id);
                    continue;
                }
                bool longShort = trade.StructureType == "多空组合";
                var positions = new List<swap_position>() { position };
                var client = DataCacheProvider.GetClientDataSource().GetData(trade.ClientId);
                SwapTradePushModel model = new SwapTradePushModel() { trade = trade, positions = positions, Client = client };
                var req = new RequestBase<SwapPosiReq>();
                req.RECORD = PrepareTradePosiReq(model, longShort);
                if (req.RECORD == null)
                {
                    pushService.DeleteLog(pushLog.id);
                    continue;
                }
                req.EXT_NO = model.extNo;
                req.SERIAL_NO = pushLog.id.ToString();
                req.ACTION = action;
                var data_req = JsonHelper.Serialize(req);
                pushService.UpdateLogReq(pushLog.id, data_req);
                kafkaProduceHelper.Produce(reqAssetSwapInsert, data_req);
            }
        }
        /// <summary>
        /// 平仓数据构造
        /// </summary>
        /// <param name="swapEvent"></param>
        /// <returns></returns>
        /// <summary>
        /// 根据平仓详细事件构造数据
        /// </summary>
        /// <param name="flowEvent"></param>
        /// <returns></returns>
        private SwapUnwindReq PrepareUnwindData(swap_event swapEvent)
        {
            SwapUnwindReq swapUnwindReq = new SwapUnwindReq();
            if (swapEvent == null)
            {
                return null;
            }
            var interestPositions = swapEvent.unwindData.FlowEvents.Where(x => x.InterestMode != (int)InterestModeEnum.初始预付金 && x.InterestMode != (int)InterestModeEnum.追加预付金).ToList();
            var floatPositions = swapEvent.unwindData.FlowEvents.Where(x => x.PayDirection > 0).ToList();
            var tradeinfFee = floatPositions.Sum(s => s.TradingFee + s.TradingFeePending);
            var markClosePnl = floatPositions.Sum(s => s.MarkClosePnl);
            swapUnwindReq.ACCTSWAP_TERMINATE.CUSTORDID = swapEvent.id.ToString();
            swapUnwindReq.ACCTSWAP_TERMINATE.TERMINATE_TYPE = swapEvent.unwindData.CloseMethod == (int)CloseMethodEnum.全部平仓 ? "1" : "0";
            swapUnwindReq.ACCTSWAP_TERMINATE.TERMINATE_COUNT = swapEvent.unwindData.CloseQty.OtcFormatMoney(false, 4);
            var unwindDate = swapEvent.unwindData.UnwindDate.HasValue ? swapEvent.unwindData.UnwindDate.Value : swapEvent.unwindData.ValueDate;
            var payDate = swapEvent.unwindData.PayDate.HasValue ? swapEvent.unwindData.PayDate.Value : swapEvent.unwindData.ValueDate;
            var fixFee = interestPositions.Sum(s => s.InterestFee * (s.InterestDirection == (int)SwapDirectionEnum.支付 ? -1 : 1));
            var gddAmount = interestPositions.Sum(s => s.InterestClosePnL) - fixFee;
            swapUnwindReq.ACCTSWAP_TERMINATE.TERMINATE_DAY = unwindDate.ToString("yyyy-MM-dd");
            swapUnwindReq.ACCTSWAP_TERMINATE.PAY_DAY = payDate.ToString("yyyy-MM-dd");
            swapUnwindReq.ACCTSWAP_TERMINATE.ZCD_AMOUNT = (markClosePnl - tradeinfFee).OtcFormatMoney(false, 4);
            swapUnwindReq.ACCTSWAP_TERMINATE.GDD_AMOUNT = gddAmount.OtcFormatMoney(false, 4);
            swapUnwindReq.ACCTSWAP_TERMINATE.ORDSTATUS = "0";
            swapUnwindReq.ACCTSWAP_TERMINATE.FIX_FEE = fixFee.OtcFormatMoney(false, 4);
            swapUnwindReq.ACCTSWAP_TERMINATE.ASSET_FEE = tradeinfFee.OtcFormatMoney(false, 4);
            return swapUnwindReq;
        }
        /// <summary>
        /// 撤单数据构造
        /// </summary>
        /// <param name="trade"></param>
        /// <param name="positions"></param>
        /// <returns></returns>
        private SwapUnwindReq PrepareRepealUnwindData(string dataId, swap_position posi, bool hasInterest)
        {
            decimal notionalValue = posi.PosiNotionalValue;
            if (posi.PosiDirection == 0)
            {
                notionalValue = posi.InterestPrincipalFix;
            }
            SwapUnwindReq swapUnwindReq = new SwapUnwindReq();
            swapUnwindReq.ACCTSWAP_TERMINATE.CUSTORDID = dataId;
            swapUnwindReq.ACCTSWAP_TERMINATE.TERMINATE_TYPE = "1";
            swapUnwindReq.ACCTSWAP_TERMINATE.TERMINATE_COUNT = posi.PosiQuantity.OtcFormatMoney(false, 4);
            swapUnwindReq.ACCTSWAP_TERMINATE.TERMINATE_DAY = DateTime.Now.ToString("yyyy-MM-dd");
            swapUnwindReq.ACCTSWAP_TERMINATE.ZCD_AMOUNT = notionalValue.OtcFormatMoney(false, 4);
            swapUnwindReq.ACCTSWAP_TERMINATE.GDD_AMOUNT = "0";
            swapUnwindReq.ACCTSWAP_TERMINATE.ORDSTATUS = "0";
            if (hasInterest)
            {
                swapUnwindReq.ACCTSWAP_TERMINATE.GDD_AMOUNT = swapUnwindReq.ACCTSWAP_TERMINATE.ZCD_AMOUNT;
            }
            return swapUnwindReq;
        }
        /// <summary>
        /// 交易请求数据组装
        /// </summary>
        /// <param name="td"></param>
        /// <param name="positions"></param>
        /// <returns></returns>
        private SwapTradeReq PrepareTradeReq(SwapTradePushModel model)
        {
            var td = model.trade;
            return PrepareTradeReqSingle(model);
        }
        /// <summary>
        /// 非多空组合交易推送
        /// </summary>
        /// <param name="model"></param>
        /// <returns></returns>
        private SwapTradeReq PrepareTradeReqSingle(SwapTradePushModel model)
        {
            var td = model.trade;
            var tradeExtend = model.tradeExtend;
            var positions = model.positions;
            var client = model.Client;
            SwapTradeReq swapTradeReq = new SwapTradeReq();
            var floatRateInterest = positions.FirstOrDefault(x => !string.IsNullOrEmpty(x.FloatRateUnderlyingCode));
            var interestList = positions.Where(x => ConsTrade.InterestModels.Contains(x.InterestMode)).ToList();
            var marginList = positions.Where(x => x.InterestMode == (int)InterestModeEnum.初始预付金).ToList();
            var marginAmount = marginList.Sum(s => s.InterestPrincipalFix * (s.InterestDirection == (int)SwapDirectionEnum.收取 ? 1 : -1));
            var position = positions.FirstOrDefault(x => x.PosiDirection > 0 && x.IsInitial);
            if (position == null)
            {
                return null;
            }
            var stockEqvNotional = position.PosiNotionalValue;
            var marginRate = stockEqvNotional == 0 ? 0 : Math.Abs(marginAmount) / Convert.ToDecimal(stockEqvNotional);
            if (interestList.Count == 0)
            {
                return null;
            }
            model.extNo = td.TradeNumber.Replace("XX", "");
            var etradeAccount = new EtradeAccountService(UserInfo).GetEtradeAccount(td.AssetBookName);
            #region BASEINFO
            // swapTradeReq.BASEINFO.PARTY_ID = client?.Number;
            swapTradeReq.BASEINFO.PARTY_NAME = td.ClientName;
            swapTradeReq.BASEINFO.TRDTYPE = GetTradetype(positions);
            swapTradeReq.BASEINFO.ORDDATE = td.TradeDate.Value.ToString("yyyy-MM-dd");
            swapTradeReq.BASEINFO.GROUP_NAME = "TRS推送-"+ td.TradeDate.Value.ToString("yyyyMMdd");
            swapTradeReq.BASEINFO.ORDSTATUS = "0";
            swapTradeReq.BASEINFO.TRADER_ID = etradeAccount?.trader_account;
            swapTradeReq.BASEINFO.SECU_INT = etradeAccount?.balance_account_trade;
            #endregion
            #region ACCTSWAP
            swapTradeReq.ACCTSWAP.PRINCIPAL = stockEqvNotional.OtcFormatMoney(false, 4);
            swapTradeReq.ACCTSWAP.PRINCIPAL_CNY = swapTradeReq.ACCTSWAP.PRINCIPAL;
            swapTradeReq.ACCTSWAP.CURRENCY = "CNY";
            swapTradeReq.ACCTSWAP.DILIVERY_CURRENCY = swapTradeReq.ACCTSWAP.CURRENCY;
            swapTradeReq.ACCTSWAP.END_DAY = td.ExerciseDate.Value.ToString("yyyy-MM-dd");

            //swapTradeReq.ACCTSWAP.CONTRACT_TYPE = "1";
            swapTradeReq.ACCTSWAP.REGULAR_TYPE = "2";
            swapTradeReq.ACCTSWAP.COUPON_TYPE = floatRateInterest != null ? "2" : "1";
            swapTradeReq.ACCTSWAP.INTEREST_BASE = "0";
            swapTradeReq.ACCTSWAP.PAYMENT_CALENDAR = new List<string>() { "CHINA_IB" };
            swapTradeReq.ACCTSWAP.DAY_COUNTER = "Actual/365 (Fixed)";
            swapTradeReq.ACCTSWAP.MARGIN_BEGIN_RATE = marginRate.OtcFormatMoney(false, 4);
            swapTradeReq.ACCTSWAP.COUPON_RATE = position.PosiGrossPrice.OtcFormatMoney(false, 10);
            swapTradeReq.ACCTSWAP.SETTLE_TYPE = "DIRECT";
            if (td.StructureType == "定义文件型债券收益互换")
            {
                swapTradeReq.ACCTSWAP.SETTLE_TYPE = "CPAS";
            }
            var settleRules = 0;
            if (floatRateInterest != null)
            {
                swapTradeReq.ACCTSWAP.IR_I_CODE = floatRateInterest.FloatRateUnderlyingCode;
                swapTradeReq.ACCTSWAP.FLOAT_METHOD = floatRateInterest.InterestType == (int)InterestTypeEnum.单利 ? "0" : "2";
            }
            if (tradeExtend != null)
            {
                swapTradeReq.ACCTSWAP.FIXED_INTEREST_RULE = GetInterestRule(tradeExtend.ExtendObj.InterestCalcMode);
                settleRules = tradeExtend.ExtendObj.SettlementRules;
            }
            swapTradeReq.ACCTSWAP.END_SETTLE_DATE = QdpCalendarHelper.GetNonHoliday(td.ExerciseDate.Value.AddDays(settleRules)).ToString("yyyy-MM-dd");
            #endregion
            #region ACCTSWAP.REGULAR_INFO
            REGULAR_INFO rEGULAR_INFO = new REGULAR_INFO();
            rEGULAR_INFO.FINAL_STUP_TYPE = "0";
            rEGULAR_INFO.PAY_FREQ = "0D";//到期支付
            rEGULAR_INFO.PAY_ADJUST = "0";
            rEGULAR_INFO.PAY_OFFSET = "1D";
            rEGULAR_INFO.RESET_FREQ = "7D";
            rEGULAR_INFO.RESET_ADJUST = "4";
            rEGULAR_INFO.INTEREST_OFFSET = "-1D";
            rEGULAR_INFO.INTEREST_ADJUST = "2";
            if (interestList.Count > 0)
            {
                var interest = interestList[0];
                var observation = DbContext.trade_obervation.FirstOrDefault(x => x.PositionId == interest.PositionId);
                if (observation != null)
                {
                    rEGULAR_INFO.INTEREST_START_DAY = observation.ObservationStart.Value.ToString("yyyy-MM-dd");
                    rEGULAR_INFO.INTEREST_END_DAY = observation.IntervalList.OrderByDescending(o => o.Date).First().Date.ToString("yyyy-MM-dd");
                }
                else
                {
                    rEGULAR_INFO.INTEREST_START_DAY = td.StartDate.Value.ToString("yyyy-MM-dd");
                    rEGULAR_INFO.INTEREST_END_DAY = td.ExerciseDate.Value.ToString("yyyy-MM-dd");
                }
                if (floatRateInterest != null)
                {
                    rEGULAR_INFO.SPREAD = floatRateInterest.InterestRateDefault.ToString();
                }
                else
                {
                    rEGULAR_INFO.RATE = interest.InterestRateDefault.ToString();
                }
            }
            swapTradeReq.ACCTSWAP.REGULAR_INFO = rEGULAR_INFO;
            #endregion
            return swapTradeReq;
        }
       
        /// <summary>
        /// 互换资产请求数据组装
        /// </summary>
        /// <param name="td"></param>
        /// <param name="positions"></param>
        /// <returns></returns>
        private SwapPosiReq PrepareTradePosiReq(SwapTradePushModel model, bool longShort)
        {
            var client = model.Client;
            model.extNo = "";
            SwapPosiReq swapPosiReq = new SwapPosiReq();
            var td = model.trade;
            var position = model.positions.Where(x => x.PosiDirection > 0).FirstOrDefault();
            if (position == null)
            {
                return null;
            }
            model.extNo = longShort ? position.PosiNumber + "-01" : position.PosiNumber;
            var underlying = DataCacheProvider.GetUnderlyingDataSource().GetData(position.UnderlyingCode);
            if (underlying == null)
            {
                return null;
            }
            decimal price = Convert.ToDecimal(underlying.Price ?? 100);
            var etradeAccount = new EtradeAccountService(UserInfo).GetEtradeAccount(td.AssetBookName);
            swapPosiReq.BASEINFO.ORDDATE = td.TradeDate.Value.ToString("yyyy-MM-dd");
            //swapPosiReq.BASEINFO.PARTY_ID = client?.Number;
            swapPosiReq.BASEINFO.PARTY_NAME = td.ClientName;
            swapPosiReq.BASEINFO.TRD_RELATED_NO = longShort ? position.PosiNumber : td.TradeNumber.Replace("XX", "");
            swapPosiReq.BASEINFO.ORDSTATUS = "0";
            swapPosiReq.BASEINFO.GROUP_NAME = "TRS推送-" + td.TradeDate.Value.ToString("yyyyMMdd");
            swapPosiReq.BASEINFO.TRDTYPE = "10";
            if ((position.PosiDirection == (int)SwapDirectionEnum.支付 && position.PositionType == (int)PositionTypeFlag.Long) || (position.PosiDirection == (int)SwapDirectionEnum.收取 && position.PositionType == (int)PositionTypeFlag.Short))
            {
                swapPosiReq.BASEINFO.TRDTYPE = "20";
            }
            // swapPosiReq.BASEINFO.SECU_INT = "TRS_YJC_ZC";
            swapPosiReq.BASEINFO.TRADER_ID = etradeAccount?.trader_account;
            swapPosiReq.BASEINFO.SECU_INT = etradeAccount?.balance_account_asset;
            #region TRADEINFO
            var underylingCode = position.UnderlyingCode.Split(".")[0];
            swapPosiReq.TRADEINFO.TRD_FEE = "0";
            swapPosiReq.TRADEINFO.SET_FEE = "0";
            swapPosiReq.TRADEINFO.I_CODE = underylingCode;
            swapPosiReq.TRADEINFO.MARKET = underlying.MarketCode;
            swapPosiReq.TRADEINFO.A_TYPE = "SPT_BD";
            swapPosiReq.TRADEINFO.PARVALUE = position.PosiQuantity.OtcFormatMoney(false, 4);
            swapPosiReq.TRADEINFO.ORDPRICE = (price * position.PosiGrossPrice).OtcFormatMoney(false, 8);
            swapPosiReq.TRADEINFO.ORDAMOUNT = (position.PosiQuantity * position.PosiGrossPrice).OtcFormatMoney(false, 4);
            swapPosiReq.TRADEINFO.OCFLAG = "0";
            swapPosiReq.TRADEINFO.SETDAYS = "1";
            #endregion
            return swapPosiReq;
        }
        private SwapPosiReq PrepareTradePosiUnwindReq(trade td, swap_flow_event flowEvent, string data_number)
        {
            SwapPosiReq swapPosiReq = new SwapPosiReq();
            if (flowEvent == null)
            {
                return null;
            }
            var underlying = DataCacheProvider.GetUnderlyingDataSource().GetData(flowEvent.UnderlyingCode);
            if (underlying == null)
            {
                return null;
            }
            decimal price = Convert.ToDecimal(underlying.Price ?? 100);
            var etradeAccount = new EtradeAccountService(UserInfo).GetEtradeAccount(td.AssetBookName);
            swapPosiReq.BASEINFO.ORDDATE = flowEvent.UnwindDate.Value.ToString("yyyy-MM-dd");
            var payDate = flowEvent.PayDate.HasValue ? flowEvent.PayDate.Value : flowEvent.UnwindDate.Value;
            swapPosiReq.BASEINFO.SETDATE = payDate.ToString("yyyy-MM-dd");
            //swapPosiReq.BASEINFO.PARTY_ID = client?.Number;
            swapPosiReq.BASEINFO.PARTY_NAME = td.ClientName;
            swapPosiReq.BASEINFO.TRD_RELATED_NO = td.TradeNumber.Replace("XX", "");
            swapPosiReq.BASEINFO.ORDSTATUS = "0";
            swapPosiReq.BASEINFO.GROUP_NAME = "TRS推送-" + flowEvent.UnwindDate.Value.ToString("yyyyMMdd");
            swapPosiReq.BASEINFO.TRDTYPE = "20";
            if ((flowEvent.PayDirection == (int)SwapDirectionEnum.支付 && flowEvent.PositionType == (int)PositionTypeFlag.Long) || (flowEvent.PayDirection == (int)SwapDirectionEnum.收取 && flowEvent.PositionType == (int)PositionTypeFlag.Short))
            {
                swapPosiReq.BASEINFO.TRDTYPE = "10";
            }
            // swapPosiReq.BASEINFO.SECU_INT = "TRS_YJC_ZC";
            swapPosiReq.BASEINFO.TRADER_ID = etradeAccount?.trader_account;
            swapPosiReq.BASEINFO.SECU_INT = etradeAccount?.balance_account_asset;
            #region TRADEINFO
            var underylingCode = flowEvent.UnderlyingCode.Split(".")[0];
            swapPosiReq.TRADEINFO.TRD_FEE = "0";
            swapPosiReq.TRADEINFO.SET_FEE = "0";
            swapPosiReq.TRADEINFO.I_CODE = underylingCode;
            swapPosiReq.TRADEINFO.MARKET = underlying.MarketCode;
            swapPosiReq.TRADEINFO.A_TYPE = "SPT_BD";
            swapPosiReq.TRADEINFO.PARVALUE = flowEvent.Quantity.OtcFormatMoney(false, 4);
            swapPosiReq.TRADEINFO.ORDPRICE = (price * flowEvent.TradingAmountAvg).OtcFormatMoney(false, 8);
            swapPosiReq.TRADEINFO.ORDAMOUNT = (flowEvent.Quantity * flowEvent.TradingAmountAvg).OtcFormatMoney(false, 4);
            swapPosiReq.TRADEINFO.OCFLAG = "1";
            swapPosiReq.TRADEINFO.SETDAYS = "0";
            #endregion
            return swapPosiReq;
        }
        private SwapMarginReq PrepareTradeMarginUnwindReq(Client client, trade td, swap_flow_event flowEvent, bool unwind, string posiNumer)
        {
            SwapMarginReq swapMarginReq = new SwapMarginReq();
            if (flowEvent == null)
            {
                return null;
            }
            var posi = DbContext.swap_position.FirstOrDefault(x => x.PosiDirection > 0 && x.IsInitial && !x.Invalid);
            if (posi == null)
            {
                return null;
            }
            var etradeAccount = new EtradeAccountService(UserInfo).GetEtradeAccount(td.AssetBookName);
            #region BASEINFO
            swapMarginReq.BASEINFO.ORDDATE = flowEvent.UnwindDate.Value.ToString("yyyy-MM-dd");
            var payDate = flowEvent.PayDate.HasValue ? flowEvent.PayDate.Value : flowEvent.UnwindDate.Value;
            swapMarginReq.BASEINFO.SETDATE = payDate.ToString("yyyy-MM-dd");
            // swapMarginReq.BASEINFO.PARTY_ID = client?.Number;
            swapMarginReq.BASEINFO.PARTY_NAME = td.ClientName;
            swapMarginReq.BASEINFO.ORDSTATUS = "0";
            swapMarginReq.BASEINFO.GROUP_NAME = "TRS推送-" + flowEvent.UnwindDate.Value.ToString("yyyyMMdd");
            swapMarginReq.BASEINFO.TRDTYPE = flowEvent.InterestDirection == (int)SwapDirectionEnum.收取 ? "128" : "127";
            swapMarginReq.BASEINFO.TRADER_ID = etradeAccount?.trader_account;
            swapMarginReq.BASEINFO.SECU_INT = etradeAccount?.balance_account_trade;
            #endregion
            #region TRADEINFO
            swapMarginReq.TRADEINFO.PARVALUE = flowEvent.InterestPrincipal.OtcFormatMoney(false, 4);
            swapMarginReq.TRADEINFO.INTEREST_AMOUNT = "0";
            #endregion
            #region RESERVE
            if (unwind)
            {
                swapMarginReq.RESERVE.IN_CUSTORDID = posiNumer;
            }
            swapMarginReq.RESERVE.CONTRACT_CODE = td.TradeNumber.Replace("XX", "");
            swapMarginReq.RESERVE.RATE = flowEvent.InterestRate.OtcFormatMoney(false, 4);
            swapMarginReq.RESERVE.DAY_COUNTER = "Actual/365 (Fixed)";
            swapMarginReq.RESERVE.CURRENCY = "CNY";
            #endregion
            return swapMarginReq;
        }
        /// <summary>
        /// 互换预付金请求数据组装
        /// </summary>
        /// <param name="model"></param>
        /// <returns></returns>
        private SwapMarginReq PrepareTradeMarginReq(SwapTradePushModel model)
        {
            model.extNo = "";
            SwapMarginReq swapMarginReq = new SwapMarginReq();
            var td = model.trade;
            var position = model.positions.FirstOrDefault();
            if (position == null)
            {
                return null;
            }
            model.extNo = position.PosiNumber;
            var etradeAccount = new EtradeAccountService(UserInfo).GetEtradeAccount(td.AssetBookName);
            #region BASEINFO
            var flowEvent = DbContext.swap_flow_event.Where(x=>x.PositionId== position.id&&x.EventType==(int)SwapFlowEventTypeEnum.开仓&&x.DataState==(int)SwapFlowDateStateEnum.完成).FirstOrDefault();
            var eventDate = td.StartDate.Value;
            var payDate = eventDate;
            if (flowEvent!=null&& flowEvent.PayDate.HasValue)
            {
                payDate=flowEvent.PayDate.Value;
            }
            swapMarginReq.BASEINFO.ORDDATE = eventDate.ToString("yyyy-MM-dd");
            swapMarginReq.BASEINFO.SETDATE = payDate.ToString("yyyy-MM-dd");
            // swapMarginReq.BASEINFO.PARTY_ID = client?.Number;
            swapMarginReq.BASEINFO.PARTY_NAME = td.ClientName;
            swapMarginReq.BASEINFO.ORDSTATUS = "0";
            swapMarginReq.BASEINFO.GROUP_NAME = "TRS推送-" + td.StartDate.Value.ToString("yyyyMMdd");
            swapMarginReq.BASEINFO.TRDTYPE = position.InterestDirection == (int)SwapDirectionEnum.收取 ? "128" : "127";
            swapMarginReq.BASEINFO.TRADER_ID = etradeAccount?.trader_account;
            swapMarginReq.BASEINFO.SECU_INT = etradeAccount?.balance_account_trade;
            #endregion
            #region TRADEINFO
            swapMarginReq.TRADEINFO.PARVALUE = position.InterestPrincipalFix.OtcFormatMoney(false, 4);
            swapMarginReq.TRADEINFO.INTEREST_AMOUNT = "0";
            #endregion
            #region RESERVE
            //swapMarginReq.RESERVE.IN_CUSTORDID = position.PosiNumber;
            swapMarginReq.RESERVE.CONTRACT_CODE = td.TradeNumber.Replace("XX", "");
            swapMarginReq.RESERVE.RATE = position.InterestRateDefault.OtcFormatMoney(false, 4);
            swapMarginReq.RESERVE.DAY_COUNTER = "Actual/365 (Fixed)";
            swapMarginReq.RESERVE.CURRENCY = position.Currency;
            #endregion
            return swapMarginReq;
        }

        /// <summary>
        /// 交易对手方请求数据组装
        /// </summary>
        /// <param name="client"></param>
        /// <returns></returns>
        private ClientPushReq PrepareClientReq(Client client)
        {
            ClientPushReq clientReq = new ClientPushReq();
            clientReq.CRM_ID = client.Number;
            clientReq.COUNTERPARTYINFO.MEMBER_LONGNAME = client.Name;
            clientReq.COUNTERPARTYINFO.SOCIAL_CREDIT_CODE = client.UnifiedSocialCreditCode;
            clientReq.COUNTERPARTYINFO.MEMBER_NAME = client.Abbreviation;
            return clientReq;
        }
        /// <summary>
        /// 交易对手方资金账户请求数据组装
        /// </summary>
        /// <param name="bankCard"></param>
        /// <param name="clientNumber"></param>
        /// <returns></returns>
        private ClientBankReq PrepareClientBankReq(List<ClientBankCard> bankCards, Client client)
        {
            ClientBankReq clientBankReq = new ClientBankReq();
            clientBankReq.CRM_ID = client.Number;
            clientBankReq.XIR_PARTY_ID = client.OutNumber;
            foreach (ClientBankCard bankCard in bankCards)
            {
                BankInfo bankInfo = new BankInfo();
                bankInfo.ACCID = bankCard.Card;
                bankInfo.CASH_ACC_NAME = bankCard.ClientName;
                bankInfo.SETTL_BANK_NAME = bankCard.Bank;
                bankInfo.PAYMENT_NUMBER = bankCard.Payment;
                clientBankReq.COUNTERPARTYTRADELIST.PARTYTRADEINFO.Add(bankInfo);
            }
            return clientBankReq;
        }

        private void PushClientData(IKafkaProduce kafkaProduceHelper, List<PushLog> pushLogs)
        {
            var pushService = new PushLogService(UserInfo);
            using var clientDb = new ClientDBContext();
            foreach (var pushLog in pushLogs)
            {
                var intId = Convert.ToInt32(pushLog.data_id);
                var client = clientDb.client.FirstOrDefault(x => x.id == intId);
                if (client == null)
                {
                    pushService.DeleteLog(pushLog.id);
                    continue;
                }
                var req = new RequestBase<ClientPushReq>();
                req.RECORD = PrepareClientReq(client);
                if (req.RECORD == null)
                {
                    pushService.DeleteLog(pushLog.id);
                    continue;
                }
                req.EXT_NO = client.Number;
                req.SERIAL_NO = pushLog.id.ToString();
                req.ACTION = "5011";
                var data_req = JsonHelper.Serialize(req);
                pushService.UpdateLogReq(pushLog.id, data_req);
                kafkaProduceHelper.Produce(reqCounterPartyInfoInsert, data_req);
            }
        }
        private void PushClientBannkData(IKafkaProduce kafkaProduceHelper, List<PushLog> pushLogs)
        {
            var pushService = new PushLogService(UserInfo);
            using var clientDb = new ClientDBContext();
            foreach (var pushLog in pushLogs)
            {
                var intId = Convert.ToInt32(pushLog.data_id);
                var client = clientDb.client.FirstOrDefault(x => x.id == intId);
                if (client == null)
                {
                    pushService.DeleteLog(pushLog.id);
                    continue;
                }
                var bankCards = clientDb.bankcard.Where(x => x.ClientId == client.id && x.ValidState != "InValid").ToList();
                var req = new RequestBase<ClientBankReq>();
                req.RECORD = PrepareClientBankReq(bankCards, client);
                if (req.RECORD == null)
                {
                    pushService.DeleteLog(pushLog.id);
                    continue;
                }
                req.EXT_NO = client.Number;
                req.SERIAL_NO = pushLog.id.ToString();
                req.ACTION = "5013";
                var data_req = JsonHelper.Serialize(req);
                pushService.UpdateLogReq(pushLog.id, data_req);
                kafkaProduceHelper.Produce(reqCounterPartyCashAccoutInsert, data_req);
            }
        }
        /// <summary>
        /// 获取交易方向
        /// </summary>
        /// <param name="positions"></param>
        /// <returns></returns>
        private string GetTradetype(List<swap_position> positions)
        {
            string direction = "151";
            var interestList = positions.Where(x => ConsTrade.InterestModels.Contains(x.InterestMode)).ToList();
            var posiList = positions.Where(x => x.PosiDirection > 0).ToList();
            if (interestList.Count == 1)
            {
                direction = interestList[0].InterestDirection == (int)SwapDirectionEnum.收取 ? "151" : "152";
                return direction;
            }
            if (posiList.Count > 0)
            {
                direction = posiList[0].PosiDirection == (int)SwapDirectionEnum.支付 ? "151" : "152";
                return direction;
            }
            return direction;
        }
        /// <summary>
        /// 获取结算利息计算方式
        /// </summary>
        /// <param name="interestCalcMode"></param>
        /// <returns></returns>
        private string GetInterestRule(string interestCalcMode)
        {
            if (interestCalcMode == "01")
            {
                return "1";
            }
            if (interestCalcMode == "10")
            {
                return "0";
            }
            if (interestCalcMode == "11")
            {
                return "2";
            }
            return "";
        }
    }
}