| | |
| | | * WebSocket 消息在回调线程中处理。下单操作参数构建等逻辑 |
| | | * 提交到独立线程池异步执行,避免阻塞 WS 回调线程。 |
| | | * |
| | | * <h3>回调设计</h3> |
| | | * 每个下单方法接受 onSuccess/onFailure 两个 Runnable。 |
| | | * 基底开仓时 onSuccess 用于标记基底已开,网格触发时通常为 null(成交状态由仓位推送驱动)。 |
| | | * |
| | | * <h3>线程模型</h3> |
| | | * <ul> |
| | | * <li><b>单线程</b>:保证下单顺序(开多→开空→止盈单),避免并发竞争</li> |
| | | * <li><b>有界队列 64</b>:防止堆积</li> |
| | | * <li><b>有界队列 64</b>:防止堆积。极端行情下最多累积 64 个任务</li> |
| | | * <li><b>CallerRunsPolicy</b>:队列满时由提交线程直接同步执行,形成自然背压</li> |
| | | * <li><b>allowCoreThreadTimeOut</b>:60s 空闲后线程回收</li> |
| | | * <li><b>allowCoreThreadTimeOut</b>:60s 空闲后线程回收,不浪费资源</li> |
| | | * </ul> |
| | | * |
| | | * <h3>调用链</h3> |
| | | * <pre> |
| | | * OkxGridTradeService.onKline → executor.openLong/openShort (基底双开 + 网格触发) |
| | | * OkxGridTradeService.onPositionUpdate → executor.placeTakeProfit (开仓成交后设止盈) |
| | | * OkxGridTradeService.stopGrid → executor.cancelAllPriceTriggeredOrders |
| | | * </pre> |
| | | * |
| | | * @author Administrator |
| | | */ |
| | | @Slf4j |
| | | public class OkxTradeExecutor { |
| | | |
| | | private final OkxConfig config; |
| | | private static final String ORDER_TYPE_CLOSE_LONG = "plan-close-long-position"; |
| | | private static final String ORDER_TYPE_CLOSE_SHORT = "plan-close-short-position"; |
| | | |
| | | private final String contract; |
| | | private final String marginMode; |
| | | private final String accountName; |
| | | |
| | | private volatile WebSocketClient wsClient; |
| | | |
| | | private final ExecutorService executor; |
| | | |
| | | public OkxTradeExecutor(OkxConfig config, String accountName) { |
| | | this.config = config; |
| | | public OkxTradeExecutor(String contract, String marginMode, String accountName) { |
| | | this.contract = contract; |
| | | this.marginMode = marginMode; |
| | | this.accountName = accountName; |
| | | this.executor = new ThreadPoolExecutor( |
| | | 1, 1, |
| | |
| | | ((ThreadPoolExecutor) executor).allowCoreThreadTimeOut(true); |
| | | } |
| | | |
| | | public void setWebSocketClient(WebSocketClient wsClient) { |
| | | this.wsClient = wsClient; |
| | | } |
| | | |
| | | public void shutdown() { |
| | | executor.shutdown(); |
| | | try { |
| | |
| | | } |
| | | } |
| | | |
| | | public void openLong(WebSocketClient wsClient, Runnable onSuccess) { |
| | | openPosition(wsClient, config.getQuantity(), OkxEnums.POSSIDE_LONG, OkxEnums.SIDE_BUY, "开多", onSuccess); |
| | | /** |
| | | * 异步市价开多。quantity 为正数(如 "1")。 |
| | | * |
| | | * @param quantity 开仓张数(正数) |
| | | * @param onSuccess 成交成功回调(可为 null) |
| | | * @param onFailure 成交失败回调(可为 null) |
| | | */ |
| | | public void openLong(String quantity, Runnable onSuccess, Runnable onFailure) { |
| | | openPosition(quantity, OkxEnums.POSSIDE_LONG, OkxEnums.SIDE_BUY, "开多", onSuccess, onFailure); |
| | | } |
| | | |
| | | public void openShort(WebSocketClient wsClient, Runnable onSuccess) { |
| | | openPosition(wsClient, config.getQuantity(), OkxEnums.POSSIDE_SHORT, OkxEnums.SIDE_SELL, "开空", onSuccess); |
| | | /** |
| | | * 异步市价开空。quantity 为正数(如 "1")。 |
| | | * |
| | | * @param quantity 开仓张数(正数) |
| | | * @param onSuccess 成交成功回调(可为 null) |
| | | * @param onFailure 成交失败回调(可为 null) |
| | | */ |
| | | public void openShort(String quantity, Runnable onSuccess, Runnable onFailure) { |
| | | openPosition(quantity, OkxEnums.POSSIDE_SHORT, OkxEnums.SIDE_SELL, "开空", onSuccess, onFailure); |
| | | } |
| | | |
| | | private void openPosition(WebSocketClient wsClient, String sz, String posSide, String side, String label, Runnable onSuccess) { |
| | | private void openPosition(String sz, String posSide, String side, String label, Runnable onSuccess, Runnable onFailure) { |
| | | executor.execute(() -> { |
| | | try { |
| | | TradeRequestParam param = buildParam(side, posSide, sz, OkxEnums.ORDTYPE_MARKET); |
| | | sendOrder(wsClient, param); |
| | | sendOrder(param); |
| | | |
| | | log.info("[TradeExec] {}已发送, 方向:{}, 数量:{}", label, posSide, sz); |
| | | if (onSuccess != null) { |
| | |
| | | } |
| | | } catch (Exception e) { |
| | | log.error("[TradeExec] {}发送失败", label, e); |
| | | if (onFailure != null) { |
| | | onFailure.run(); |
| | | } |
| | | } |
| | | }); |
| | | } |
| | | |
| | | public void placeTakeProfit(WebSocketClient wsClient, String posSide, BigDecimal triggerPrice, String size) { |
| | | /** |
| | | * 异步创建止盈条件单(仓位计划止盈止损)。 |
| | | * |
| | | * <p>通过 OKX WebSocket 发送 algo order(conditional order),服务器监控价格, |
| | | * 达到触发价后自动平指定张数。 |
| | | * |
| | | * <h3>orderType 说明</h3> |
| | | * <ul> |
| | | * <li>plan-close-long-position:平多仓,posSide=long, side=sell</li> |
| | | * <li>plan-close-short-position:平空仓,posSide=short, side=buy</li> |
| | | * </ul> |
| | | * |
| | | * <p>止盈单创建失败时,立即市价平仓兜底(marketClose)。 |
| | | * |
| | | * @param triggerPrice 触发价格 |
| | | * @param orderType stop 类型(plan-close-long-position / plan-close-short-position) |
| | | * @param size 平仓张数(正数) |
| | | */ |
| | | public void placeTakeProfit(BigDecimal triggerPrice, String orderType, String size) { |
| | | executor.execute(() -> { |
| | | try { |
| | | String side = OkxEnums.POSSIDE_LONG.equals(posSide) ? OkxEnums.SIDE_SELL : OkxEnums.SIDE_BUY; |
| | | String posSide; |
| | | String side; |
| | | if (ORDER_TYPE_CLOSE_LONG.equals(orderType)) { |
| | | posSide = OkxEnums.POSSIDE_LONG; |
| | | side = OkxEnums.SIDE_SELL; |
| | | } else { |
| | | posSide = OkxEnums.POSSIDE_SHORT; |
| | | side = OkxEnums.SIDE_BUY; |
| | | } |
| | | |
| | | try { |
| | | TradeRequestParam param = buildParam(side, posSide, size, OkxEnums.ORDTYPE_LIMIT); |
| | | param.setMarkPx(triggerPrice.toString()); |
| | | |
| | | List<TradeRequestParam> params = new ArrayList<>(); |
| | | params.add(param); |
| | | sendBatchOrders(wsClient, params); |
| | | log.info("[TradeExec] 止盈单已发送, 方向:{}, 触发价:{}, 数量:{}", posSide, triggerPrice, size); |
| | | sendBatchOrders(params); |
| | | log.info("[TradeExec] 止盈单已发送, 触发价:{}, 类型:{}, size:{}", triggerPrice, orderType, size); |
| | | } catch (Exception e) { |
| | | log.error("[TradeExec] 止盈单发送失败", e); |
| | | log.error("[TradeExec] 止盈单发送失败, 触发价:{}, size:{}, 立即市价止盈", triggerPrice, size, e); |
| | | marketClose(side, posSide, size); |
| | | } |
| | | }); |
| | | } |
| | | |
| | | /** |
| | | * 市价止盈:在止盈条件单创建失败时立即市价平仓。 |
| | | */ |
| | | private void marketClose(String side, String posSide, String size) { |
| | | try { |
| | | TradeRequestParam param = buildParam(side, posSide, size, OkxEnums.ORDTYPE_MARKET); |
| | | param.setTradeType("3"); |
| | | sendOrder(param); |
| | | log.info("[TradeExec] 市价止盈已发送, posSide:{}, size:{}", posSide, size); |
| | | } catch (Exception e) { |
| | | log.error("[TradeExec] 市价止盈也失败, posSide:{}, size:{}", posSide, size, e); |
| | | } |
| | | } |
| | | |
| | | /** |
| | | * 异步清除指定合约的所有止盈止损条件单。 |
| | | */ |
| | | public void cancelAllPriceTriggeredOrders() { |
| | | executor.execute(() -> { |
| | | try { |
| | | if (wsClient == null || !wsClient.isOpen()) { |
| | | log.warn("[TradeExec] WS未连接,跳过撤销条件单"); |
| | | return; |
| | | } |
| | | JSONArray argsArray = new JSONArray(); |
| | | JSONObject args = new JSONObject(); |
| | | args.put("instId", contract); |
| | | args.put("algoOrdType", "conditional"); |
| | | argsArray.add(args); |
| | | |
| | | String connId = OkxWsUtil.getOrderNum("cancel"); |
| | | JSONObject msg = OkxWsUtil.buildJsonObject(connId, "cancel-algos", argsArray); |
| | | wsClient.send(msg.toJSONString()); |
| | | log.info("[TradeExec] 已发送撤销所有条件单请求"); |
| | | } catch (Exception e) { |
| | | log.error("[TradeExec] 撤销条件单失败", e); |
| | | } |
| | | }); |
| | | } |
| | |
| | | private TradeRequestParam buildParam(String side, String posSide, String sz, String ordType) { |
| | | TradeRequestParam param = new TradeRequestParam(); |
| | | param.setAccountName(accountName); |
| | | param.setInstId(config.getContract()); |
| | | param.setTdMode(config.getMarginMode()); |
| | | param.setInstId(contract); |
| | | param.setTdMode(marginMode); |
| | | param.setPosSide(posSide); |
| | | param.setOrdType(ordType); |
| | | param.setSide(side); |
| | |
| | | return param; |
| | | } |
| | | |
| | | private void sendOrder(WebSocketClient wsClient, TradeRequestParam param) { |
| | | private void sendOrder(TradeRequestParam param) { |
| | | if (wsClient == null || !wsClient.isOpen()) { |
| | | log.warn("[TradeExec] WS未连接,跳过下单"); |
| | | return; |
| | |
| | | log.info("[TradeExec] 发送下单: side={}, sz={}", param.getSide(), param.getSz()); |
| | | } |
| | | |
| | | private void sendBatchOrders(WebSocketClient wsClient, List<TradeRequestParam> params) { |
| | | private void sendBatchOrders(List<TradeRequestParam> params) { |
| | | if (wsClient == null || !wsClient.isOpen() || params == null || params.isEmpty()) { |
| | | log.warn("[TradeExec] WS未连接或参数为空,跳过批量下单"); |
| | | return; |