From c37093473edbac5a0bd9af129339e60790057946 Mon Sep 17 00:00:00 2001 From: Codex Date: Sat, 8 Aug 2026 02:59:24 +0900 Subject: [PATCH] feat: execute full basic catalog safely --- README.md | 10 + .../candidate/OrderCandidateBatchAdapter.java | 18 +- .../worker/runtime/EvaluatingBotRuntime.java | 177 ++++++++++++++++-- .../EvaluationRuntimeConfiguration.java | 11 +- .../PostgresExecutionGateStateSource.java | 81 ++++++++ .../OrderCandidateBatchAdapterTest.java | 17 +- .../worker/runtime/EvaluationLoopE2ETest.java | 12 ++ .../worker/runtime/ExecutionGateTest.java | 60 ++++++ .../runtime/plan/BasicPlanInterpreter.java | 121 +++++++++++- .../plan/BasicPlanInterpreterTest.java | 35 ++++ .../domain/candidate/CandidateOrder.java | 26 ++- .../messaging/evaluation/OrderCandidate.java | 40 +++- .../evaluation/OrderCandidateBatch.java | 3 + .../PostgresScopedCandidateComposition.java | 5 +- 14 files changed, 586 insertions(+), 30 deletions(-) create mode 100644 apps/trading-worker/src/main/java/com/idea2strategy/trading/worker/runtime/PostgresExecutionGateStateSource.java create mode 100644 apps/trading-worker/src/test/java/com/idea2strategy/trading/worker/runtime/ExecutionGateTest.java diff --git a/README.md b/README.md index 6af0f5f..70d3daf 100644 --- a/README.md +++ b/README.md @@ -39,3 +39,13 @@ gradle :apps:trading-worker:bootRun 두 App 모두 Docker 내부 네트워크에서 동작하며 호스트 포트를 열지 않습니다. +## Basic catalog execution + +`basic-elements:2026-08-08`로 발행된 전략은 `30m`, `1h`, `4h`, `1d` 중 하나의 주기만 +사용합니다. 전체 Basic 조건 카탈로그를 실시간 상태로 평가하며, 주문 블록의 주문 비율, +1회 실행, 주기 실행, 조건 재충족, N봉·N거래일 대기, 최대 실행 횟수를 bot별 순차 gate에서 +결정적으로 적용합니다. 매수 비율은 equal-allocation share에, 매도 비율은 composer가 읽은 +실제 보유 포지션에 적용됩니다. worker 재시작 시에는 현재 포지션 주기의 canonical order +intent에서 실행 횟수와 마지막 실행 시각을 복원하며, 포지션이 완전히 종료된 뒤 다음 주기가 +시작될 때 gate를 초기화합니다. 이전 catalog reader는 이미 릴리스된 bot의 동작 보존용입니다. + diff --git a/apps/trading-worker/src/main/java/com/idea2strategy/trading/worker/candidate/OrderCandidateBatchAdapter.java b/apps/trading-worker/src/main/java/com/idea2strategy/trading/worker/candidate/OrderCandidateBatchAdapter.java index 9095c94..5256da9 100644 --- a/apps/trading-worker/src/main/java/com/idea2strategy/trading/worker/candidate/OrderCandidateBatchAdapter.java +++ b/apps/trading-worker/src/main/java/com/idea2strategy/trading/worker/candidate/OrderCandidateBatchAdapter.java @@ -30,7 +30,7 @@ public final class OrderCandidateBatchAdapter { public CandidateBatch toDomain(OrderCandidateBatch source) { int version = source.schemaVersion(); - if (version < MINIMUM_SCHEMA_VERSION || version > OrderCandidateBatch.ALLOCATION_SCHEMA_VERSION) { + if (version < MINIMUM_SCHEMA_VERSION || version > OrderCandidateBatch.PARTIAL_POSITION_SCHEMA_VERSION) { throw new IllegalArgumentException( "Unsupported order candidate batch schema version: " + version); } @@ -76,6 +76,19 @@ private static void requireShapeOfVersion(int version, OrderCandidate candidate) "schema version " + version + " BUY candidate " + candidate.candidateId() + " must carry an allocation share"); } + if (version < OrderCandidateBatch.PARTIAL_POSITION_SCHEMA_VERSION + && candidate.requestedPositionPercent().isPresent()) { + throw new IllegalArgumentException( + "schema version " + version + " candidate " + candidate.candidateId() + + " must not carry a position percentage"); + } + if (version >= OrderCandidateBatch.PARTIAL_POSITION_SCHEMA_VERSION + && candidate.side() == OrderSide.SELL + && candidate.requestedPositionPercent().isEmpty()) { + throw new IllegalArgumentException( + "schema version " + version + " SELL candidate " + candidate.candidateId() + + " must carry a position percentage"); + } } private CandidateOrder toDomain(OrderCandidate source) { @@ -90,6 +103,7 @@ private CandidateOrder toDomain(OrderCandidate source) { .orElse(null), source.referencePrice(), source.limitPrice(), - source.reasonCodes()); + source.reasonCodes(), + source.positionPercent()); } } diff --git a/apps/trading-worker/src/main/java/com/idea2strategy/trading/worker/runtime/EvaluatingBotRuntime.java b/apps/trading-worker/src/main/java/com/idea2strategy/trading/worker/runtime/EvaluatingBotRuntime.java index 5ffdaf0..cde1303 100644 --- a/apps/trading-worker/src/main/java/com/idea2strategy/trading/worker/runtime/EvaluatingBotRuntime.java +++ b/apps/trading-worker/src/main/java/com/idea2strategy/trading/worker/runtime/EvaluatingBotRuntime.java @@ -44,6 +44,7 @@ import java.util.Optional; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Supplier; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -86,6 +87,7 @@ public final class EvaluatingBotRuntime implements BotRuntimeLifecycle { private final BotScopeResolver scopeResolver; private final EvaluationRunRecorder runRecorder; private final PositionMetricSource positionMetricSource; + private final ExecutionGateStateSource executionGateStateSource; private final BasicPlanInterpreter interpreter = new BasicPlanInterpreter(); private final BasicStrategyExecutor executor = new BasicStrategyExecutor(); private final BasicCandidateConverger converger = new BasicCandidateConverger(); @@ -96,7 +98,8 @@ public EvaluatingBotRuntime( OrderCandidateBatchAdapter adapter, BotScopeResolver scopeResolver, EvaluationRunRecorder runRecorder) { - this(processor, adapter, scopeResolver, runRecorder, PositionMetricSource.none()); + this(processor, adapter, scopeResolver, runRecorder, + PositionMetricSource.none(), ExecutionGateStateSource.none()); } public EvaluatingBotRuntime( @@ -105,11 +108,24 @@ public EvaluatingBotRuntime( BotScopeResolver scopeResolver, EvaluationRunRecorder runRecorder, PositionMetricSource positionMetricSource) { + this(processor, adapter, scopeResolver, runRecorder, + positionMetricSource, ExecutionGateStateSource.none()); + } + + public EvaluatingBotRuntime( + CandidateBatchProcessor processor, + OrderCandidateBatchAdapter adapter, + BotScopeResolver scopeResolver, + EvaluationRunRecorder runRecorder, + PositionMetricSource positionMetricSource, + ExecutionGateStateSource executionGateStateSource) { this.processor = Objects.requireNonNull(processor, "processor"); this.adapter = Objects.requireNonNull(adapter, "adapter"); this.scopeResolver = Objects.requireNonNull(scopeResolver, "scopeResolver"); this.runRecorder = Objects.requireNonNull(runRecorder, "runRecorder"); this.positionMetricSource = Objects.requireNonNull(positionMetricSource, "positionMetricSource"); + this.executionGateStateSource = + Objects.requireNonNull(executionGateStateSource, "executionGateStateSource"); } @Override @@ -196,7 +212,7 @@ private Optional evaluate( evaluationId, bot.plan().flows(), inputsFor(bot, event, snapshot, price, marketValues))); BasicCandidateConvergenceResult converged = - converger.converge(evaluationId, acceptedOf(execution, evaluationId)); + converger.converge(evaluationId, acceptedOf(bot, execution, evaluationId, event)); if (converged.acceptedCandidates().isEmpty()) { return Optional.empty(); } @@ -256,18 +272,36 @@ private Map inputsFor( return inputs; } - private List acceptedOf(BasicExecutionResult execution, UUID evaluationId) { + private List acceptedOf( + RegisteredBot bot, + BasicExecutionResult execution, + UUID evaluationId, + MarketEventEnvelope event) { List candidates = new ArrayList<>(); execution.decisions().stream() - .filter(decision -> decision.status() == BasicDecisionStatus.CANDIDATE) - .forEach(decision -> candidates.add(new BasicOrderCandidate( + .filter(decision -> decision.instrumentId().equals(event.instrumentId())) + .forEach(decision -> { + BasicPlanInterpreter.ExecutionPolicy policy = + bot.plan().executionPolicyByFlowKey().get(decision.flowId()); + if (!bot.executionGate( + decision.flowId(), + decision.instrumentId(), + () -> executionGateStateSource.resolve( + bot.botId(), decision.flowId(), decision.instrumentId())) + .accepts(decision.status(), policy, event.occurredAt())) { + return; + } + candidates.add(new BasicOrderCandidate( derived("candidate", evaluationId + ":" + decision.flowId() + ":" + decision.instrumentId()), decision.flowId(), decision.instrumentId(), decision.side(), decision.buyAllocation(), - Map.of("evaluationId", evaluationId.toString())))); + Map.of( + "evaluationId", evaluationId.toString(), + "orderPercent", Integer.toString(policy.orderPercent())))); + }); return candidates; } @@ -288,18 +322,19 @@ private OrderCandidateBatch batchOf( BigDecimal referencePrice) { List candidates = new ArrayList<>(); for (BasicOrderCandidate candidate : converged.acceptedCandidates()) { + int orderPercent = Integer.parseInt(candidate.actionParameters().get("orderPercent")); candidates.add(candidate.side() == BasicOrderSide.BUY ? OrderCandidate.allocatedBuy( candidate.candidateId(), candidate.instrumentId(), scope.flowId(), - candidate.buyAllocation().orElseThrow().numerator(), - candidate.buyAllocation().orElseThrow().denominator(), + candidate.buyAllocation().orElseThrow().numerator() * orderPercent, + candidate.buyAllocation().orElseThrow().denominator() * 100, referencePrice, null, List.of("BASIC_RULE_MATCHED")) - : OrderCandidate.heldSell( + : OrderCandidate.partialHeldSell( candidate.candidateId(), candidate.instrumentId(), scope.flowId(), - referencePrice, null, List.of("BASIC_RULE_MATCHED"))); + orderPercent, referencePrice, null, List.of("BASIC_RULE_MATCHED"))); } return new OrderCandidateBatch( - OrderCandidateBatch.ALLOCATION_SCHEMA_VERSION, + OrderCandidateBatch.PARTIAL_POSITION_SCHEMA_VERSION, derived("candidate-batch", bot.botId() + ":" + event.eventId()), evaluationId, bot.botId(), @@ -366,6 +401,7 @@ private static final class RegisteredBot { private final EvaluationWindow window; private final Map signalStates = new LinkedHashMap<>(); private final Map positionTrackers = new LinkedHashMap<>(); + private final Map executionGates = new LinkedHashMap<>(); /** The gateway's stream position, which starts wherever the bot joined. */ private long lastMarketSequence = Long.MIN_VALUE; @@ -428,12 +464,100 @@ private BasicMarketSignalState signalState(UUID instrumentId) { } private PositionTracker positionTracker(UUID instrumentId, PositionSnapshot snapshot) { - return positionTrackers.compute(instrumentId, (ignored, current) -> - current != null && current.matches(snapshot) ? current : new PositionTracker(snapshot)); + PositionTracker current = positionTrackers.get(instrumentId); + if (current != null && current.matches(snapshot)) { + return current; + } + clearExecutionGates(instrumentId); + PositionTracker replacement = new PositionTracker(snapshot); + positionTrackers.put(instrumentId, replacement); + return replacement; } private void clearPositionTracker(UUID instrumentId) { - positionTrackers.remove(instrumentId); + if (positionTrackers.remove(instrumentId) != null) { + clearExecutionGates(instrumentId); + } + } + + private ExecutionGate executionGate( + String flowId, UUID instrumentId, Supplier snapshot) { + return executionGates.computeIfAbsent( + flowId + ":" + instrumentId, ignored -> new ExecutionGate(snapshot.get())); + } + + private void clearExecutionGates(UUID instrumentId) { + String suffix = ":" + instrumentId; + executionGates.keySet().removeIf(key -> key.endsWith(suffix)); + } + } + + static final class ExecutionGate { + private int executions; + private int barsSinceExecution; + private Instant lastExecutionAt; + private boolean conditionRearmed = true; + + ExecutionGate() { + this(ExecutionGateSnapshot.empty()); + } + + ExecutionGate(ExecutionGateSnapshot snapshot) { + Objects.requireNonNull(snapshot, "snapshot"); + executions = snapshot.executions(); + lastExecutionAt = snapshot.lastExecutionAt(); + conditionRearmed = executions == 0; + } + + boolean accepts( + BasicDecisionStatus status, + BasicPlanInterpreter.ExecutionPolicy policy, + Instant occurredAt) { + if (lastExecutionAt != null) { + barsSinceExecution++; + } + if (status != BasicDecisionStatus.CANDIDATE) { + if (status == BasicDecisionStatus.CONDITION_NOT_MET) { + conditionRearmed = true; + } + return false; + } + int limit = policy.executionMode().equals("1회만") + ? 1 + : policy.maxExecutions(); + if (executions >= limit) { + return false; + } + boolean eligible = executions == 0 + || policy.executionMode().equals("주기마다") + || switch (policy.waitMode()) { + case "조건 재충족" -> conditionRearmed; + case "N봉 이후" -> barsSinceExecution >= policy.waitInterval(); + case "N거래일 이후" -> tradingDaysSince(lastExecutionAt, occurredAt) + >= policy.waitInterval(); + default -> false; + }; + if (!eligible) { + return false; + } + executions++; + barsSinceExecution = 0; + lastExecutionAt = occurredAt; + conditionRearmed = false; + return true; + } + + private static long tradingDaysSince(Instant start, Instant end) { + LocalDate cursor = start.atZone(ZoneId.of("America/New_York")).toLocalDate(); + LocalDate through = end.atZone(ZoneId.of("America/New_York")).toLocalDate(); + long days = 0; + while (cursor.isBefore(through)) { + cursor = cursor.plusDays(1); + if (cursor.getDayOfWeek().getValue() <= 5) { + days++; + } + } + return days; } } @@ -541,6 +665,31 @@ static PositionMetricSource none() { } } + /** Restores a flow's execution limiter from canonical intents in the current position cycle. */ + public interface ExecutionGateStateSource { + ExecutionGateSnapshot resolve(UUID botId, String flowKey, UUID instrumentId); + + static ExecutionGateStateSource none() { + return (botId, flowKey, instrumentId) -> ExecutionGateSnapshot.empty(); + } + } + + public record ExecutionGateSnapshot(int executions, Instant lastExecutionAt) { + public ExecutionGateSnapshot { + if (executions < 0) { + throw new IllegalArgumentException("executions must not be negative"); + } + if ((executions == 0) != (lastExecutionAt == null)) { + throw new IllegalArgumentException( + "lastExecutionAt must exist exactly when executions are present"); + } + } + + static ExecutionGateSnapshot empty() { + return new ExecutionGateSnapshot(0, null); + } + } + public record PositionSnapshot(BigDecimal averageEntryPrice, Instant openedAt) { public PositionSnapshot { Objects.requireNonNull(averageEntryPrice, "averageEntryPrice"); diff --git a/apps/trading-worker/src/main/java/com/idea2strategy/trading/worker/runtime/EvaluationRuntimeConfiguration.java b/apps/trading-worker/src/main/java/com/idea2strategy/trading/worker/runtime/EvaluationRuntimeConfiguration.java index 85b6ef9..627583f 100644 --- a/apps/trading-worker/src/main/java/com/idea2strategy/trading/worker/runtime/EvaluationRuntimeConfiguration.java +++ b/apps/trading-worker/src/main/java/com/idea2strategy/trading/worker/runtime/EvaluationRuntimeConfiguration.java @@ -34,14 +34,21 @@ PostgresPositionMetricSource positionMetricSource(JdbcClient jdbc) { return new PostgresPositionMetricSource(jdbc); } + @Bean + PostgresExecutionGateStateSource executionGateStateSource(JdbcClient jdbc) { + return new PostgresExecutionGateStateSource(jdbc); + } + @Bean EvaluatingBotRuntime evaluatingBotRuntime( CandidateBatchProcessor processor, OrderCandidateBatchAdapter adapter, PostgresBotScopeResolver scopeResolver, PostgresEvaluationRunRecorder runRecorder, - PostgresPositionMetricSource positionMetricSource) { + PostgresPositionMetricSource positionMetricSource, + PostgresExecutionGateStateSource executionGateStateSource) { return new EvaluatingBotRuntime( - processor, adapter, scopeResolver, runRecorder, positionMetricSource); + processor, adapter, scopeResolver, runRecorder, + positionMetricSource, executionGateStateSource); } } diff --git a/apps/trading-worker/src/main/java/com/idea2strategy/trading/worker/runtime/PostgresExecutionGateStateSource.java b/apps/trading-worker/src/main/java/com/idea2strategy/trading/worker/runtime/PostgresExecutionGateStateSource.java new file mode 100644 index 0000000..a645a79 --- /dev/null +++ b/apps/trading-worker/src/main/java/com/idea2strategy/trading/worker/runtime/PostgresExecutionGateStateSource.java @@ -0,0 +1,81 @@ +package com.idea2strategy.trading.worker.runtime; + +import java.time.OffsetDateTime; +import java.util.Objects; +import java.util.UUID; +import org.springframework.jdbc.core.simple.JdbcClient; + +/** Restores execution limits from immutable intents emitted in the current position cycle. */ +final class PostgresExecutionGateStateSource + implements EvaluatingBotRuntime.ExecutionGateStateSource { + + /** + * A position cycle starts after the most recently closed long lot, or when the bot started if it + * has never closed one. Counting intent rows deliberately matches the live gate: it limits + * condition-qualified order attempts even when downstream composition rejects the attempt. + */ + private static final String RESOLVE = """ + with cycle as ( + select greatest( + coalesce(max(projection.closed_at), '-infinity'::timestamptz), + coalesce(bot.started_at, bot.execution_eligible_from) + ) as started_at + from bot.bots bot + left join trading.position_lots lot + on lot.bot_id = bot.id + and lot.instrument_id = :instrumentId + and cast(lot.lot_side as varchar) = 'LONG' + left join trading.position_lot_projections projection + on projection.position_lot_id = lot.id + and projection.closed_at is not null + where bot.id = :botId + group by bot.started_at, bot.execution_eligible_from + ) + select count(intent.id) as executions, + max(run.queued_at) as last_execution_at + from cycle + join bot.bot_partitions partition on partition.bot_id = :botId + join bot.flows flow + on flow.partition_id = partition.id + and flow.name = :flowKey + left join trading.order_intents intent + on intent.bot_id = :botId + and intent.flow_id = flow.id + and intent.instrument_id = :instrumentId + and cast(intent.origin_type as varchar) = 'FLOW_EVALUATION' + left join bot.evaluation_runs run + on run.bot_id = intent.bot_id + and run.id = intent.evaluation_run_id + and run.queued_at >= cycle.started_at + where intent.id is null or run.id is not null + """; + + private final JdbcClient jdbc; + + PostgresExecutionGateStateSource(JdbcClient jdbc) { + this.jdbc = Objects.requireNonNull(jdbc, "jdbc"); + } + + @Override + public EvaluatingBotRuntime.ExecutionGateSnapshot resolve( + UUID botId, String flowKey, UUID instrumentId) { + Objects.requireNonNull(botId, "botId"); + Objects.requireNonNull(flowKey, "flowKey"); + Objects.requireNonNull(instrumentId, "instrumentId"); + return jdbc.sql(RESOLVE) + .param("botId", botId) + .param("flowKey", flowKey) + .param("instrumentId", instrumentId) + .query((resultSet, rowNumber) -> { + int executions = resultSet.getInt("executions"); + OffsetDateTime last = + resultSet.getObject("last_execution_at", OffsetDateTime.class); + return executions == 0 + ? EvaluatingBotRuntime.ExecutionGateSnapshot.empty() + : new EvaluatingBotRuntime.ExecutionGateSnapshot( + executions, last.toInstant()); + }) + .optional() + .orElseGet(EvaluatingBotRuntime.ExecutionGateSnapshot::empty); + } +} diff --git a/apps/trading-worker/src/test/java/com/idea2strategy/trading/worker/candidate/OrderCandidateBatchAdapterTest.java b/apps/trading-worker/src/test/java/com/idea2strategy/trading/worker/candidate/OrderCandidateBatchAdapterTest.java index c8a10bc..1f9da33 100644 --- a/apps/trading-worker/src/test/java/com/idea2strategy/trading/worker/candidate/OrderCandidateBatchAdapterTest.java +++ b/apps/trading-worker/src/test/java/com/idea2strategy/trading/worker/candidate/OrderCandidateBatchAdapterTest.java @@ -92,7 +92,7 @@ void rejectsUnsupportedSchemaVersion() { // A version beyond the allocation one is refused even though it carries a complete scope, // because its meaning has not been agreed with the producer yet. OrderCandidateBatch source = new OrderCandidateBatch( - OrderCandidateBatch.ALLOCATION_SCHEMA_VERSION + 1, + OrderCandidateBatch.PARTIAL_POSITION_SCHEMA_VERSION + 1, UUID.fromString("10000000-0000-0000-0000-000000000001"), UUID.fromString("20000000-0000-0000-0000-000000000002"), UUID.fromString("50000000-0000-0000-0000-000000000005"), @@ -104,6 +104,21 @@ void rejectsUnsupportedSchemaVersion() { assertThrows(IllegalArgumentException.class, () -> new OrderCandidateBatchAdapter().toDomain(source)); } + @Test + void carriesTheRequestedSellPercentageOfASchemaVersionFourCandidate() { + OrderCandidate candidate = OrderCandidate.partialHeldSell( + CANDIDATE, INSTRUMENT, FLOW, 40, null, null, List.of("EXIT")); + OrderCandidateBatch batch = new OrderCandidateBatch( + OrderCandidateBatch.PARTIAL_POSITION_SCHEMA_VERSION, + BATCH, EVALUATION, BOT, PARTITION, SOURCE_EVENT, + Instant.parse("2026-08-01T00:00:00Z"), List.of(candidate)); + + var translated = new OrderCandidateBatchAdapter().toDomain(batch) + .candidates().getFirst(); + + assertEquals(40, translated.requestedPositionPercent().orElseThrow()); + } + /** A version 3 buy hands over its share and lets this service size it. */ @Test void carriesTheAllocationShareOfASchemaVersionThreeBuy() { diff --git a/apps/trading-worker/src/test/java/com/idea2strategy/trading/worker/runtime/EvaluationLoopE2ETest.java b/apps/trading-worker/src/test/java/com/idea2strategy/trading/worker/runtime/EvaluationLoopE2ETest.java index 3e2173e..0e72d94 100644 --- a/apps/trading-worker/src/test/java/com/idea2strategy/trading/worker/runtime/EvaluationLoopE2ETest.java +++ b/apps/trading-worker/src/test/java/com/idea2strategy/trading/worker/runtime/EvaluationLoopE2ETest.java @@ -154,6 +154,18 @@ void aStartedBotTurnsAMarketEventIntoACanonicalOrder() { "select count(*) from trading.resource_reservations where bot_id = ?", BOT))); } + @Test + void workerRestartRestoresTheOneShotExecutionLimitFromCanonicalIntents() { + runtime.start(plan(), warmup(), EvaluationWindow.openEndedFrom(ELIGIBLE_FROM)); + assertEquals(1, runtime.feed(event(1, "84")).size()); + + runtime.stop(BOT, "WORKER_RESTART"); + runtime.start(plan(), warmup(), EvaluationWindow.openEndedFrom(ELIGIBLE_FROM)); + + assertTrue(runtime.feed(eventAt(2, "83", EVENT_AT.plusSeconds(60))).isEmpty()); + assertEquals(1, count("select count(*) from trading.order_intents where bot_id = ?", BOT)); + } + @Test void aDirectPriceBlockTradesFromCompletedMarketBarsWithoutRsi() { runtime.start(directPricePlan(), PreparedWarmup.none(), diff --git a/apps/trading-worker/src/test/java/com/idea2strategy/trading/worker/runtime/ExecutionGateTest.java b/apps/trading-worker/src/test/java/com/idea2strategy/trading/worker/runtime/ExecutionGateTest.java new file mode 100644 index 0000000..c81eeb7 --- /dev/null +++ b/apps/trading-worker/src/test/java/com/idea2strategy/trading/worker/runtime/ExecutionGateTest.java @@ -0,0 +1,60 @@ +package com.idea2strategy.trading.worker.runtime; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import com.idea2strategy.trading.strategy.runtime.basic.BasicDecisionStatus; +import com.idea2strategy.trading.strategy.runtime.plan.BasicPlanInterpreter; +import java.time.Instant; +import org.junit.jupiter.api.Test; + +class ExecutionGateTest { + + private static final Instant FIRST = Instant.parse("2026-08-03T14:00:00Z"); + + @Test + void oneShotAndConditionRearmModesEnforceTheirExecutionBoundaries() { + var once = new EvaluatingBotRuntime.ExecutionGate(); + var rearm = new EvaluatingBotRuntime.ExecutionGate(); + var oncePolicy = new BasicPlanInterpreter.ExecutionPolicy( + 100, "1회만", "조건 재충족", 1, 10); + var rearmPolicy = new BasicPlanInterpreter.ExecutionPolicy( + 100, "대기 후 재진입", "조건 재충족", 1, 2); + + assertTrue(once.accepts(BasicDecisionStatus.CANDIDATE, oncePolicy, FIRST)); + assertFalse(once.accepts(BasicDecisionStatus.CANDIDATE, oncePolicy, FIRST.plusSeconds(60))); + + assertTrue(rearm.accepts(BasicDecisionStatus.CANDIDATE, rearmPolicy, FIRST)); + assertFalse(rearm.accepts(BasicDecisionStatus.CANDIDATE, rearmPolicy, FIRST.plusSeconds(60))); + assertFalse(rearm.accepts( + BasicDecisionStatus.CONDITION_NOT_MET, rearmPolicy, FIRST.plusSeconds(120))); + assertTrue(rearm.accepts(BasicDecisionStatus.CANDIDATE, rearmPolicy, FIRST.plusSeconds(180))); + assertFalse(rearm.accepts(BasicDecisionStatus.CANDIDATE, rearmPolicy, FIRST.plusSeconds(240))); + } + + @Test + void barWaitCountsCompletedEvaluationsBeforeReentry() { + var gate = new EvaluatingBotRuntime.ExecutionGate(); + var policy = new BasicPlanInterpreter.ExecutionPolicy( + 25, "대기 후 재진입", "N봉 이후", 2, 3); + + assertTrue(gate.accepts(BasicDecisionStatus.CANDIDATE, policy, FIRST)); + assertFalse(gate.accepts(BasicDecisionStatus.CANDIDATE, policy, FIRST.plusSeconds(60))); + assertTrue(gate.accepts(BasicDecisionStatus.CANDIDATE, policy, FIRST.plusSeconds(120))); + } + + @Test + void restoredGateFailsClosedUntilItsWaitConditionIsObservedAgain() { + var gate = new EvaluatingBotRuntime.ExecutionGate( + new EvaluatingBotRuntime.ExecutionGateSnapshot(1, FIRST)); + var policy = new BasicPlanInterpreter.ExecutionPolicy( + 50, "대기 후 재실행", "조건 재충족", 1, 3); + + assertFalse(gate.accepts( + BasicDecisionStatus.CANDIDATE, policy, FIRST.plusSeconds(30))); + assertFalse(gate.accepts( + BasicDecisionStatus.CONDITION_NOT_MET, policy, FIRST.plusSeconds(60))); + assertTrue(gate.accepts( + BasicDecisionStatus.CANDIDATE, policy, FIRST.plusSeconds(90))); + } +} diff --git a/modules/strategy-runtime/src/main/java/com/idea2strategy/trading/strategy/runtime/plan/BasicPlanInterpreter.java b/modules/strategy-runtime/src/main/java/com/idea2strategy/trading/strategy/runtime/plan/BasicPlanInterpreter.java index 524afa8..7a3421e 100644 --- a/modules/strategy-runtime/src/main/java/com/idea2strategy/trading/strategy/runtime/plan/BasicPlanInterpreter.java +++ b/modules/strategy-runtime/src/main/java/com/idea2strategy/trading/strategy/runtime/plan/BasicPlanInterpreter.java @@ -19,6 +19,7 @@ import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Set; import java.util.UUID; /** @@ -70,6 +71,11 @@ public final class BasicPlanInterpreter { private static final String LOAD_FEATURE = "LOAD_FEATURE"; private static final String COMPARE = "COMPARE"; private static final String EMIT_ORDER_CANDIDATE = "EMIT_ORDER_CANDIDATE"; + private static final String PRODUCTION_CATALOG_VERSION = "basic-elements:2026-08-08"; + private static final Set PRODUCTION_RESOLUTIONS = Set.of("30m", "1h", "4h", "1d"); + private static final Set EXECUTION_MODES = Set.of( + "1회만", "주기마다", "대기 후 재진입", "대기 후 재실행"); + private static final Set WAIT_MODES = Set.of("조건 재충족", "N봉 이후", "N거래일 이후"); private static final MathContext MATH = new MathContext(18, RoundingMode.HALF_UP); private final ObjectMapper objectMapper = new ObjectMapper(); @@ -91,10 +97,16 @@ public InterpretedPlan interpret(String planDocument) { + PLAN_SCHEMA_VERSION + " nor " + MULTI_CONTAINER_PLAN_SCHEMA_VERSION); } - CompiledContainer planWide = perFlow ? null : container(root, "plan"); + JsonNode catalogVersionNode = root.get("elementCatalogVersion"); + boolean productionCatalog = catalogVersionNode != null + && catalogVersionNode.isTextual() + && PRODUCTION_CATALOG_VERSION.equals(catalogVersionNode.asText()); + CompiledContainer planWide = perFlow ? null : container(root, "plan", productionCatalog); List flows = new ArrayList<>(); Map partitionKeyByFlowKey = new LinkedHashMap<>(); + Map executionPolicyByFlowKey = new LinkedHashMap<>(); + java.util.Set productionResolutions = new java.util.LinkedHashSet<>(); JsonNode partitions = object(root, "executionSnapshot").get("partitions"); if (partitions == null || !partitions.isArray() || partitions.isEmpty()) { throw reject("executionSnapshot.partitions must be a non-empty array"); @@ -114,9 +126,23 @@ public InterpretedPlan interpret(String planDocument) { } instrumentNodes.forEach(node -> instruments.add(UUID.fromString(node.asText()))); CompiledContainer container = - perFlow ? container(flowNode, "flow " + flowKey) : planWide; + perFlow ? container(flowNode, "flow " + flowKey, productionCatalog) : planWide; flows.add(new BasicFlow( flowKey, container.side(), instruments, container.conditionSteps())); + executionPolicyByFlowKey.put(flowKey, container.executionPolicy()); + if (productionCatalog) { + container.conditionStepsSource().stream() + .map(PlanStep::arguments) + .filter(Objects::nonNull) + .map(arguments -> arguments.get("resolution")) + .filter(Objects::nonNull) + .forEach(value -> { + if (!value.isTextual() || !PRODUCTION_RESOLUTIONS.contains(value.asText())) { + throw reject("production resolution must be one of " + PRODUCTION_RESOLUTIONS); + } + productionResolutions.add(value.asText()); + }); + } if (partitionKeyByFlowKey.put(flowKey, partitionKey) != null) { throw reject("flow key " + flowKey + " is declared more than once"); } @@ -125,7 +151,11 @@ public InterpretedPlan interpret(String planDocument) { if (flows.isEmpty()) { throw reject("a compiled plan declares no flows"); } - return new InterpretedPlan(flows, Map.copyOf(partitionKeyByFlowKey)); + if (productionResolutions.size() > 1) { + throw reject("a production plan must use one resolution across all flows"); + } + return new InterpretedPlan( + flows, Map.copyOf(partitionKeyByFlowKey), Map.copyOf(executionPolicyByFlowKey)); } /** @@ -136,7 +166,7 @@ public InterpretedPlan interpret(String planDocument) { * {@code EMIT_ORDER_CANDIDATE} is not a condition — it is where the side and the allocation are * declared — so it is consumed here rather than evaluated per instrument. */ - private CompiledContainer container(JsonNode owner, String description) { + private CompiledContainer container(JsonNode owner, String description, boolean productionCatalog) { List steps = steps(owner); PlanStep terminal = steps.getLast(); if (!EMIT_ORDER_CANDIDATE.equals(terminal.operation())) { @@ -150,14 +180,78 @@ private CompiledContainer container(JsonNode owner, String description) { for (PlanStep step : conditionSteps) { compiled.add(new BasicConditionStep(step.stepId(), evaluatorFor(step))); } + BasicOrderSide side = BasicOrderSide.valueOf(argument(terminal, "side")); + ExecutionPolicy executionPolicy = productionCatalog + ? new ExecutionPolicy( + decimalPercent(terminal, "orderPercent"), + argument(terminal, "executionMode"), + argument(terminal, "waitMode"), + positiveInteger(terminal, "waitInterval"), + positiveInteger(terminal, "maxExecutions")) + : ExecutionPolicy.legacy(); + if (productionCatalog && !validExecutionMode(side, executionPolicy.executionMode())) { + throw reject("executionMode " + executionPolicy.executionMode() + " is not valid for " + side); + } return new CompiledContainer( - BasicOrderSide.valueOf(argument(terminal, "side")), + side, argument(terminal, "allocation"), - List.copyOf(compiled)); + List.copyOf(compiled), List.copyOf(conditionSteps), executionPolicy); + } + + private static boolean validExecutionMode(BasicOrderSide side, String mode) { + return side == BasicOrderSide.BUY + ? Set.of("1회만", "주기마다", "대기 후 재진입").contains(mode) + : Set.of("1회만", "대기 후 재실행").contains(mode); } private record CompiledContainer( - BasicOrderSide side, String allocation, List conditionSteps) {} + BasicOrderSide side, + String allocation, + List conditionSteps, + List conditionStepsSource, + ExecutionPolicy executionPolicy) {} + + public record ExecutionPolicy( + int orderPercent, + String executionMode, + String waitMode, + int waitInterval, + int maxExecutions) { + + public ExecutionPolicy { + if (orderPercent < 1 || orderPercent > 100) { + throw reject("orderPercent must be between 1 and 100"); + } + if (waitInterval < 1 || maxExecutions < 1) { + throw reject("waitInterval and maxExecutions must be positive integers"); + } + executionMode = Objects.requireNonNull(executionMode, "executionMode"); + waitMode = Objects.requireNonNull(waitMode, "waitMode"); + if (!EXECUTION_MODES.contains(executionMode) || !WAIT_MODES.contains(waitMode)) { + throw reject("executionMode or waitMode is not supported"); + } + } + + static ExecutionPolicy legacy() { + return new ExecutionPolicy(100, "1회만", "조건 재충족", 1, 1); + } + } + + private static int decimalPercent(PlanStep step, String name) { + try { + return new BigDecimal(argument(step, name)).intValueExact(); + } catch (ArithmeticException exception) { + throw reject(name + " must be a whole percent"); + } + } + + private static int positiveInteger(PlanStep step, String name) { + try { + return Integer.parseInt(argument(step, name)); + } catch (NumberFormatException exception) { + throw reject(name + " must be an integer"); + } + } /** * The evaluator for one plan step. @@ -714,15 +808,26 @@ private static IllegalArgumentException reject(String detail) { */ public record InterpretedPlan( List flows, - Map partitionKeyByFlowKey) { + Map partitionKeyByFlowKey, + Map executionPolicyByFlowKey) { + + public InterpretedPlan(List flows, Map partitionKeyByFlowKey) { + this(flows, partitionKeyByFlowKey, flows.stream().collect(java.util.stream.Collectors.toUnmodifiableMap( + BasicFlow::flowId, ignored -> ExecutionPolicy.legacy()))); + } public InterpretedPlan { flows = List.copyOf(Objects.requireNonNull(flows, "flows")); partitionKeyByFlowKey = Map.copyOf( Objects.requireNonNull(partitionKeyByFlowKey, "partitionKeyByFlowKey")); + executionPolicyByFlowKey = Map.copyOf( + Objects.requireNonNull(executionPolicyByFlowKey, "executionPolicyByFlowKey")); if (flows.isEmpty()) { throw new IllegalArgumentException("an interpreted plan carries at least one flow"); } + if (!executionPolicyByFlowKey.keySet().equals(partitionKeyByFlowKey.keySet())) { + throw new IllegalArgumentException("every interpreted flow needs one execution policy"); + } } /** Every instrument any flow of this plan evaluates. */ diff --git a/modules/strategy-runtime/src/test/java/com/idea2strategy/trading/strategy/runtime/plan/BasicPlanInterpreterTest.java b/modules/strategy-runtime/src/test/java/com/idea2strategy/trading/strategy/runtime/plan/BasicPlanInterpreterTest.java index 8eb9733..84302d7 100644 --- a/modules/strategy-runtime/src/test/java/com/idea2strategy/trading/strategy/runtime/plan/BasicPlanInterpreterTest.java +++ b/modules/strategy-runtime/src/test/java/com/idea2strategy/trading/strategy/runtime/plan/BasicPlanInterpreterTest.java @@ -218,6 +218,22 @@ void aDirectBarBlockWaitsForTheRequestedBarToClose() { result.decisions().getFirst().trace().getFirst().reasonCode())); } + @Test + void productionCatalogAcceptsOnlyNewResolutionsAndCarriesTerminalExecutionPolicy() { + String accepted = productionDirectPlan("4h"); + String legacyResolution = productionDirectPlan("1m"); + + var plan = interpreter.interpret(accepted); + + assertAll( + () -> assertEquals(25, plan.executionPolicyByFlowKey().get("flow-1").orderPercent()), + () -> assertEquals("대기 후 재진입", + plan.executionPolicyByFlowKey().get("flow-1").executionMode()), + () -> assertEquals(3, plan.executionPolicyByFlowKey().get("flow-1").maxExecutions()), + () -> assertThrows(IllegalArgumentException.class, + () -> interpreter.interpret(legacyResolution))); + } + // ------------------------------------------------------------------ fixtures private BasicExecutionResult execute( @@ -268,6 +284,25 @@ private static String directPlan(String operation, String arguments, String side """.formatted(INSTRUMENT, operation, arguments, side); } + private static String productionDirectPlan(String resolution) { + return directPlan( + "PRICE_COMPARE", + "\"resolution\":\"" + resolution + + "\",\"operator\":\"GT\",\"reference\":\"PREVIOUS_CLOSE\"", + "BUY") + .replace( + "{\"schemaVersion\":\"basic-compiled-plan.v2\"", + "{\"schemaVersion\":\"basic-compiled-plan.v2\"," + + "\"elementCatalogVersion\":\"basic-elements:2026-08-08\"") + .replace( + "\"allocation\":\"EQUAL\",\"orderType\":\"MARKET\",\"side\":\"BUY\"", + "\"allocation\":\"EQUAL\",\"orderType\":\"MARKET\"," + + "\"timeInForce\":\"DAY\",\"side\":\"BUY\"," + + "\"orderPercent\":\"25\",\"executionMode\":\"대기 후 재진입\"," + + "\"waitMode\":\"N봉 이후\",\"waitInterval\":\"2\"," + + "\"maxExecutions\":\"3\""); + } + /** * Root #202: the ordinary Basic strategy — a buy container and a sell container over the same * instrument, each an AND chain of its own blocks. diff --git a/modules/trading-domain/src/main/java/com/idea2strategy/trading/domain/candidate/CandidateOrder.java b/modules/trading-domain/src/main/java/com/idea2strategy/trading/domain/candidate/CandidateOrder.java index cf4a96c..4ffab47 100644 --- a/modules/trading-domain/src/main/java/com/idea2strategy/trading/domain/candidate/CandidateOrder.java +++ b/modules/trading-domain/src/main/java/com/idea2strategy/trading/domain/candidate/CandidateOrder.java @@ -29,7 +29,8 @@ public record CandidateOrder( CandidateAllocation allocation, BigDecimal referencePrice, BigDecimal limitPrice, - List reasonCodes) { + List reasonCodes, + Integer positionPercent) { public CandidateOrder { candidateId = Objects.requireNonNull(candidateId, "candidateId"); @@ -52,9 +53,28 @@ public record CandidateOrder( + "position held"); } } + if (positionPercent != null + && (!"SELL".equals(side) || positionPercent < 1 || positionPercent > 100)) { + throw new IllegalArgumentException( + "positionPercent is allowed only for SELL and must be between 1 and 100"); + } reasonCodes = List.copyOf(Objects.requireNonNull(reasonCodes, "reasonCodes")); } + public CandidateOrder( + UUID candidateId, + UUID instrumentId, + UUID flowId, + String side, + BigDecimal quantity, + CandidateAllocation allocation, + BigDecimal referencePrice, + BigDecimal limitPrice, + List reasonCodes) { + this(candidateId, instrumentId, flowId, side, quantity, allocation, referencePrice, + limitPrice, reasonCodes, null); + } + /** The unscoped shape, kept for candidates that arrive on schema version 1. */ public CandidateOrder( UUID candidateId, UUID instrumentId, String side, BigDecimal quantity, @@ -82,4 +102,8 @@ public Optional requestedQuantity() { public Optional allocationShare() { return Optional.ofNullable(allocation); } + + public Optional requestedPositionPercent() { + return Optional.ofNullable(positionPercent); + } } diff --git a/modules/trading-messaging/src/main/java/com/idea2strategy/trading/messaging/evaluation/OrderCandidate.java b/modules/trading-messaging/src/main/java/com/idea2strategy/trading/messaging/evaluation/OrderCandidate.java index 88f277d..692c432 100644 --- a/modules/trading-messaging/src/main/java/com/idea2strategy/trading/messaging/evaluation/OrderCandidate.java +++ b/modules/trading-messaging/src/main/java/com/idea2strategy/trading/messaging/evaluation/OrderCandidate.java @@ -41,7 +41,8 @@ public record OrderCandidate( Integer allocationDenominator, BigDecimal referencePrice, BigDecimal limitPrice, - List reasonCodes) { + List reasonCodes, + Integer positionPercent) { public OrderCandidate { candidateId = Objects.requireNonNull(candidateId, "candidateId"); @@ -57,9 +58,29 @@ public record OrderCandidate( referencePrice = requirePositive(referencePrice, "referencePrice"); } requireAllocation(side, quantity, allocationNumerator, allocationDenominator); + if (positionPercent != null + && (side != OrderSide.SELL || positionPercent < 1 || positionPercent > 100)) { + throw new IllegalArgumentException( + "positionPercent is allowed only for SELL and must be between 1 and 100"); + } reasonCodes = List.copyOf(Objects.requireNonNull(reasonCodes, "reasonCodes")); } + public OrderCandidate( + UUID candidateId, + UUID instrumentId, + UUID flowId, + OrderSide side, + BigDecimal quantity, + Integer allocationNumerator, + Integer allocationDenominator, + BigDecimal referencePrice, + BigDecimal limitPrice, + List reasonCodes) { + this(candidateId, instrumentId, flowId, side, quantity, allocationNumerator, + allocationDenominator, referencePrice, limitPrice, reasonCodes, null); + } + /** The version 1 shape, kept so an existing producer keeps deserialising unchanged. */ public OrderCandidate( UUID candidateId, @@ -118,6 +139,23 @@ public static OrderCandidate heldSell( referencePrice, limitPrice, reasonCodes); } + public static OrderCandidate partialHeldSell( + UUID candidateId, + UUID instrumentId, + UUID flowId, + int positionPercent, + BigDecimal referencePrice, + BigDecimal limitPrice, + List reasonCodes) { + return new OrderCandidate( + candidateId, instrumentId, flowId, OrderSide.SELL, null, null, null, + referencePrice, limitPrice, reasonCodes, positionPercent); + } + + public Optional requestedPositionPercent() { + return Optional.ofNullable(positionPercent); + } + /** The owning flow, present from schema version 2. */ public Optional flow() { return Optional.ofNullable(flowId); diff --git a/modules/trading-messaging/src/main/java/com/idea2strategy/trading/messaging/evaluation/OrderCandidateBatch.java b/modules/trading-messaging/src/main/java/com/idea2strategy/trading/messaging/evaluation/OrderCandidateBatch.java index a1bbc55..273eb00 100644 --- a/modules/trading-messaging/src/main/java/com/idea2strategy/trading/messaging/evaluation/OrderCandidateBatch.java +++ b/modules/trading-messaging/src/main/java/com/idea2strategy/trading/messaging/evaluation/OrderCandidateBatch.java @@ -38,6 +38,9 @@ public record OrderCandidateBatch( */ public static final int ALLOCATION_SCHEMA_VERSION = 3; + /** The first schema version that can request a percentage of a held sell position. */ + public static final int PARTIAL_POSITION_SCHEMA_VERSION = 4; + public OrderCandidateBatch { if (schemaVersion < 1) { throw new IllegalArgumentException("schemaVersion must be positive"); diff --git a/modules/trading-persistence/src/main/java/com/idea2strategy/trading/persistence/candidate/PostgresScopedCandidateComposition.java b/modules/trading-persistence/src/main/java/com/idea2strategy/trading/persistence/candidate/PostgresScopedCandidateComposition.java index bc9eca6..8a8d5cf 100644 --- a/modules/trading-persistence/src/main/java/com/idea2strategy/trading/persistence/candidate/PostgresScopedCandidateComposition.java +++ b/modules/trading-persistence/src/main/java/com/idea2strategy/trading/persistence/candidate/PostgresScopedCandidateComposition.java @@ -320,7 +320,10 @@ private Sizing size( BigDecimal available = lots.stream() .map(LotReservationAllocation::reservedQuantity) .reduce(BigDecimal.ZERO, BigDecimal::add); - BigDecimal wanted = candidate.requestedQuantity().orElse(available); + BigDecimal wanted = candidate.requestedQuantity().orElseGet(() -> + available.multiply(BigDecimal.valueOf( + candidate.requestedPositionPercent().orElse(100))) + .divide(BigDecimal.valueOf(100), 0, RoundingMode.DOWN)); if (available.signum() == 0 || wanted.signum() == 0) { Sizing rejected = new Sizing( side, candidate.requestedQuantity().orElse(MINIMUM_REQUESTED_QUANTITY));