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(); private IKafkaProduce kafkaProduceHelper; public SwapConsumerService(OptUserInfo optUser) : base(optUser) { } public void SetKafKaProduce(IKafkaProduce kafkaProduce) { kafkaProduceHelper = kafkaProduce; } /// /// Trs对冲交易接口 /// /// public void ConsumerTrsExchangeTradeResp(KafkaConsumerHelper _kafkaConsumer, string tradeSource) { _kafkaConsumer.Subscribe(msg => { LogFactory.GetLogger("SwapConsumerService").Info("TRS对冲交易回执接口:"+ msg); if (!string.IsNullOrEmpty(msg)) { var result = JsonHelper.Deserialize(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); } } }); } /// /// 接入对冲标的 /// /// 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.Info("同步场内交易失败,数据格式不正确:" + 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); } /// /// Trs成交流水接口 /// /// public void ConsumerTrsSwapFlowResp(KafkaConsumerHelper _kafkaConsumer) { _kafkaConsumer.Subscribe(msg => { if (!string.IsNullOrEmpty(msg)) { var result = JsonHelper.Deserialize(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); } } }); } /// /// 校验标的信息 /// /// /// /// 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; } } }