package com.xcong.excoin.quartz.job; import cn.hutool.core.util.StrUtil; import com.alibaba.fastjson.JSON; import com.huobi.client.SubscriptionClient; import com.huobi.client.SubscriptionOptions; import com.huobi.client.model.Candlestick; import com.huobi.client.model.enums.CandlestickInterval; import com.xcong.excoin.modules.coin.dao.OrderCoinDealDao; import com.xcong.excoin.modules.coin.dao.OrderCoinsDao; import com.xcong.excoin.modules.coin.entity.OrderCoinsEntity; import com.xcong.excoin.modules.coin.service.OrderCoinService; import com.xcong.excoin.modules.symbols.constants.SymbolsConstats; import com.xcong.excoin.modules.symbols.service.SymbolsService; import com.xcong.excoin.processor.CoinProcessor; import com.xcong.excoin.processor.CoinProcessorFactory; import com.xcong.excoin.processor.DefaultCoinProcessor; import com.xcong.excoin.processor.MarketService; import com.xcong.excoin.rabbit.pricequeue.WebsocketPriceService; import com.xcong.excoin.rabbit.producer.ExchangeProducer; import com.xcong.excoin.trade.CoinTrader; import com.xcong.excoin.trade.CoinTraderFactory; import com.xcong.excoin.trade.ExchangeTrade; import com.xcong.excoin.utils.CoinTypeConvert; import com.xcong.excoin.utils.RedisUtils; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.stereotype.Component; import javax.annotation.PostConstruct; import javax.annotation.Resource; import java.math.BigDecimal; import java.text.ParseException; import java.util.ArrayList; import java.util.List; import java.util.Map; /** * 开启撮合交易 * * @author wzy * @date 2020-05-28 **/ @Slf4j @Component @ConditionalOnProperty(prefix = "app", name = "exchange-trade", havingValue = "true") public class CoinTradeInitJob { @Resource private OrderCoinsDao orderCoinsDao; @Resource private OrderCoinDealDao orderCoinDealDao; @Resource private CoinTraderFactory factory; @Resource private RedisUtils redisUtils; @Resource private MarketService marketService; @Resource private CoinProcessorFactory processorFactory; @Resource ExchangeProducer exchangeProducer; @PostConstruct public void initCoinTrade() { log.info("#=======撮合交易器开启=======#"); String symbol = SymbolsConstats.CPV; CoinTrader newTrader = new CoinTrader(symbol); newTrader.setExchangeProducer(exchangeProducer); //newTrader.setKafkaTemplate(kafkaTemplate); //newTrader.setBaseCoinScale(coin.getBaseCoinScale()); //newTrader.setCoinScale(coin.getCoinScale()); // newTrader.setPublishType(coin.getPublishType()); //newTrader.setClearTime(coin.getClearTime()); // 创建成功以后需要对未处理订单预处理 log.info("======CoinTrader Process: " + symbol + "======"); List symbolList = new ArrayList<>(); symbolList.add(symbol); List orders = orderCoinsDao.selectCoinOrderOnTrade(symbolList); List tradingOrders = new ArrayList<>(); List completedOrders = new ArrayList<>(); orders.forEach(order -> { tradingOrders.add(order); }); try { newTrader.trade(tradingOrders); } catch (ParseException e) { e.printStackTrace(); log.info("异常:trader.trade(tradingOrders);"); } newTrader.setReady(true); factory.addTrader(symbol, newTrader); // 创建K线生成器 CoinProcessor processor = new DefaultCoinProcessor(symbol, "USDT"); processor.setMarketService(marketService); processor.setRedisUtils(redisUtils); //processor.setExchangeRate(exchangeRate); processor.initializeThumb(); //processor.initializeUsdRate(); processor.setIsHalt(false); List nekk = orderCoinDealDao.selectOrderCoinDealByTime(SymbolsConstats.CPV, null, null); processor.process(nekk); String symbolUsdt = symbol; if(!symbol.contains("USDT")){ symbolUsdt = symbol+"/USDT"; } String key = "NEW_KINE_{}"; key = StrUtil.format(key, symbolUsdt); Object o = redisUtils.get(key); if(o!=null){ Map currentKlineMap = (Map )o; ((DefaultCoinProcessor) processor).setCurrentKlineMap(currentKlineMap); } processorFactory.addProcessor(symbol, processor); } }