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.common.contants.AppContants; 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.HashMap; 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.ROC; 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 -> { if(order.getDealCnt()==null){ order.setDealCnt(BigDecimal.ZERO); } if(order.getDealAmount()==null){ order.setDealAmount(BigDecimal.ZERO); } 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.ROC, 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); }else{ // 当最新K线不存在时 需要初始化 // 1min 5min 15min 30min 1hour 4hour 1day 1week String[] rang = {"1min","5min","15min","30min","1hour","4hour","1day","1week"}; Map currentKlineMap = new HashMap<>(); long currentTimeMillis = System.currentTimeMillis(); for (String s : rang) { Candlestick candlestick = new Candlestick(); candlestick.setClose(AppContants.DEFAULT_PRICE); candlestick.setHigh(AppContants.DEFAULT_PRICE); candlestick.setLow(AppContants.DEFAULT_PRICE); candlestick.setOpen(AppContants.DEFAULT_PRICE); candlestick.setTimestamp(currentTimeMillis); currentKlineMap.put(s,candlestick); } redisUtils.set(key,currentKlineMap); } processorFactory.addProcessor(symbol, processor); } }