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 =>
            {
                var hedgingSource = OtcAppConfigHelper.HedgingSource==1|| OtcAppConfigHelper.HedgingSource==0;
                if (hedgingSource)
                {
                    return;
                }
                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(SwapPosiReq swapPosiReq, string tradeSource)
        {
            var underlying = CheckUnderlying(swapPosiReq.TRADEINFO.I_CODE, swapPosiReq.TRADEINFO.MARKET);
            var orddDate = TryFormatDate(swapPosiReq.BASEINFO.ORDDATE);
            var longType = swapPosiReq.BASEINFO.TRDTYPE == "10" ? "多头" : "空头";
            var openType = swapPosiReq.TRADEINFO.OCFLAG == "0" ? "开仓" : "平仓";

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

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


        /// <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>
        /// 校验标的信息
        /// </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;
        }
    }
}