zszq-trs/YLErpDAL/Modules/SwapModule/SwapConsumerService.cs

using BaseOUDAL;
using Confluent.Kafka;
using DocumentFormat.OpenXml.Wordprocessing;
using Microsoft.Office.Interop.Excel;
using CsvHelper;
using Newtonsoft.Json;
using Qdp.Foundation.Utilities;
using YLErp.Abstract;
using YLErp.BLL;
using YLErp.BLL.Eod;
using YLErp.Helpers;
using YLErp.Model.HengTaiModel;
using YLErp.Modules.ExchangeTradeModule;

namespace YLErp.Modules.SwapModule
{
    public class SwapConsumerService : YLBaseService
    {

        static readonly IYcLogger logger = LogFactory.GetLogger<SwapConsumerService>();

        private IKafkaProduce kafkaProduceHelper;
        public SwapConsumerService(OptUserInfo optUser) : base(optUser)
        {
        }
        public void SetKafKaProduce(IKafkaProduce kafkaProduce)
        {
            kafkaProduceHelper = kafkaProduce;
        }


        /// <summary>
        /// Trs对冲交易接口
        /// </summary>
        /// <param name="_kafkaConsumer"></param>
        public void ConsumerTrsExchangeTradeResp(KafkaConsumerHelper _kafkaConsumer, string tradeSource)
        {
            _kafkaConsumer.Subscribe(msg =>
            {
                LogFactory.GetLogger("SwapConsumerService").Info("TRS对冲交易回执接口:"+ msg);
                if (!string.IsNullOrEmpty(msg))
                {
                    var result = JsonHelper.Deserialize<SwapAsset>(msg);
                    if (result == null)
                    {
                        return;
                    }
                    try
                    {
                        var account = result.SECU_INT;
                        var HTAccounts = DbContext.exchange_account.Where(x => !string.IsNullOrEmpty(x.AccountCode)).Select(s => s.AccountCode).Distinct().ToList();
                        if (HTAccounts.Contains(account))
                        {
                            PrepareExchangeTradeData(result, tradeSource);
                            Task.Run(() =>
                            {
                                RealtimePnlCalc.RealtimeSwapPosition(new OptUserInfo(0, "互换实时持仓服务", OptUserFrom.Service));
                            });
                            return;
                        }
                    }
                    catch (Exception e)
                    {
                        LogFactory.GetLogger("SwapConsumerService").Error("TRS对冲交易回执接口", e);
                    }

                }
            });
        }


        /// <summary>
        /// 接入对冲标的
        /// </summary>
        /// <param name="swapPosiReq"></param>
        private void PrepareExchangeTradeData(SwapAsset swapPosiReq, string tradeSource)
        {
            var underlying = CheckUnderlying(swapPosiReq.I_CODE, swapPosiReq.MARKET);
            var orddDate = TryFormatDate(swapPosiReq.ORD_DATE);
            var longType = swapPosiReq.TRD_TYPE == "10" ? "多头" : "空头";
            var openType = swapPosiReq.OC_FLAG == "0" ? "开仓" : "平仓";

            if (string.IsNullOrEmpty(swapPosiReq.EXT_NO) || !double.TryParse(swapPosiReq.PAR_VALUE, out _) || !double.TryParse(swapPosiReq.ORD_PRICE, out _))
            {
                logger.Error("同步场内交易失败，数据格式不正确：" + JsonHelper.Serialize(swapPosiReq));
                return;
            }

            ExchangeTradeSaveApiReq req = new ExchangeTradeSaveApiReq
            {
                ExchangeAccountCode = swapPosiReq.SECU_INT,
                BuySell = longType + openType,
                Commission = 0,
                IsHistory = orddDate.Date < valuedateBLL.ValueDate ? true : false,
                Lots = double.Parse(swapPosiReq.PAR_VALUE) / underlying.ContractSize,
                TradeDate = orddDate.Date,
                TradeNumber = swapPosiReq.EXT_NO,
                TradePrice = double.Parse(swapPosiReq.ORD_PRICE) / (underlying.Price ?? 100),
                TradeType = underlying.UnderlyingInstrumentTypeCn,
                UnderlyingCode = underlying.UnderlyingCode
            };
            var service = new ExchangeTradeSyncApiService(UserInfo);
            service.Save(req, tradeSource);
        }

        /// <summary>
        /// Trs成交流水接口
        /// </summary>
        /// <param name="_kafkaConsumer"></param>
        public void ConsumerTrsSwapFlowResp(KafkaConsumerHelper _kafkaConsumer)
        {
            _kafkaConsumer.Subscribe(msg =>
            {
                if (!string.IsNullOrEmpty(msg))
                {
                    var result = JsonHelper.Deserialize<swap_flow>(msg);
                    if (result == null)
                    {
                        return;
                    }
                    try
                    {
                        var swapFlow = DbContext.swap_flow.Where(x => x.trs_deal_id == result.trs_deal_id).FirstOrDefault();
                        if (swapFlow == null)
                        {
                            swapFlow = new swap_flow();
                        }
                        swapFlow.BsType = result.BsType;
                        swapFlow.ClientId = result.ClientId;
                        swapFlow.ClientName = result.ClientName;
                        swapFlow.ContractSize = result.ContractSize;
                        swapFlow.DataState = (int)SwapFlowDateStateEnum.等待完成;
                        swapFlow.DealType = 1;
                        swapFlow.OccurTime = result.OccurTime;
                        swapFlow.OptId = result.OptId;
                        swapFlow.OptName = result.OptName;
                        swapFlow.OptTime = result.OptTime;
                        swapFlow.SettleDate = result.SettleDate;
                        swapFlow.TradingAmount = result.TradingAmount;
                        swapFlow.TradingAmountAvg = result.TradingAmountAvg;
                        swapFlow.TradingAmountFeeAvg = result.TradingAmountFeeAvg;
                        swapFlow.TradingAmountNet = result.TradingAmountNet;
                        swapFlow.TradingAmountNetFee = result.TradingAmountNetFee;
                        swapFlow.TradingFee =Math.Round( result.TradingFee, ConsGlobal.MoneyRound, MidpointRounding.AwayFromZero);
                        swapFlow.TradingQty = result.TradingQty;
                        swapFlow.trs_deal_id = result.trs_deal_id;
                        swapFlow.UnderlyingCode = result.UnderlyingCode;
                        swapFlow.UnderlyingName = result.UnderlyingName;
                        swapFlow.ytm = result.ytm;
                        if (swapFlow.id == 0)
                        {
                            DbContext.swap_flow.Add(swapFlow);
                        }
                        DbContext.SaveChanges();
                        RealtimePnlCalc.RealtimeSwapPosition(new OptUserInfo(0, "互换实时持仓服务", OptUserFrom.Service));
                    }
                    catch (Exception e)
                    {
                        LogFactory.GetLogger("SwapConsumerService").Error("TRS成交流水接口", e);
                    }

                }
            });
        }
        /// <summary>
        /// 校验标的信息
        /// </summary>
        /// <param name="underlyingCode"></param>
        /// <returns></returns>
        /// <exception cref="ServiceException"></exception>
        private underlying_manager CheckUnderlying(string underlyingCode, string market)
        {
            var underlying = DbContext.underlying_manager.FirstOrDefault(x => x.UnderlyingCode.StartsWith(underlyingCode + ".") && x.MarketCode == market);
            if (underlying == null)
            {
                throw new ServiceException($"标的代码{underlyingCode}不存在");
            }
            return underlying;
        }
        private DateTime TryFormatDate(string val)
        {
            if (!DateTime.TryParse(val, out DateTime formatDate))
            {
                throw new ServiceException($"参数{val}不是正确的日期格式");
            }
            return formatDate;
        }
    }
}