quant-trader-service/src/main/java/com/quantai/trader/persistence/JdbcTraderDecisionTraceWriter.java

package com.quantai.trader.persistence;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.quantai.trader.config.TraderProperties;
import com.quantai.trader.domain.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Repository;

import java.sql.Timestamp;
import java.util.Map;

@Repository
public class JdbcTraderDecisionTraceWriter implements TraderDecisionTraceWriter {
    private static final Logger log = LoggerFactory.getLogger(JdbcTraderDecisionTraceWriter.class);

    private final JdbcTemplate jdbcTemplate;
    private final TraderJsonCodec jsonCodec;
    private final TraderProperties properties;

    public JdbcTraderDecisionTraceWriter(JdbcTemplate jdbcTemplate, ObjectMapper objectMapper, TraderProperties properties) {
        this.jdbcTemplate = jdbcTemplate;
        this.jsonCodec = new TraderJsonCodec(objectMapper);
        this.properties = properties;
    }

    @Override
    public void persistCycleTrace(TraderDecisionCycle cycle,
                                  TraderMarketSnapshot snapshot,
                                  TraderModelOutput modelOutput,
                                  TraderPositionState positionState,
                                  TraderPositionManagerDecision pmDecision,
                                  TraderRiskDecision riskDecision,
                                  TraderAction action) {
        upsertRun(cycle);
        insertCycle(cycle, positionState, riskDecision);
        insertModelOutput(modelOutput, snapshot);
        insertPmDecision(pmDecision);
        insertRiskDecision(riskDecision);
        insertAction(action);
        log.info("event=trader.trace.persisted runId={} cycleId={} action={} riskAllowed={}",
                cycle.runId(), cycle.cycleId(), action.actionType(), riskDecision.allowAction());
    }

    void upsertRun(TraderDecisionCycle cycle) {
        jdbcTemplate.update("""
                        insert into trader_run
                          (run_id, run_mode, symbol, model_bundle_version, calibration_bundle_version,
                           pm_config_version, execution_mode, status, config_json, started_at)
                        values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                        on duplicate key update
                          model_bundle_version = values(model_bundle_version),
                          calibration_bundle_version = values(calibration_bundle_version),
                          pm_config_version = values(pm_config_version),
                          execution_mode = values(execution_mode),
                          status = values(status),
                          updated_at = current_timestamp(3)
                        """,
                cycle.runId(), cycle.runMode().name(), cycle.symbol(), cycle.modelBundleVersion(),
                cycle.calibrationBundleVersion(), cycle.pmConfigVersion(), properties.execution().mode().name(),
                "RUNNING", jsonCodec.toJson(Map.of(
                        "serviceName", properties.serviceName(),
                        "artifactRoot", properties.artifact().artifactRoot(),
                        "redisKeyPrefix", properties.runtime().redisKeyPrefix())),
                Timestamp.from(cycle.cycleTime()));
    }

    void insertCycle(TraderDecisionCycle cycle, TraderPositionState positionState, TraderRiskDecision riskDecision) {
        jdbcTemplate.update("""
                        insert into trader_decision_cycle
                          (run_id, cycle_id, symbol, cycle_time, state, decision_status, blocker)
                        values (?, ?, ?, ?, ?, ?, ?)
                        """,
                cycle.runId(), cycle.cycleId(), cycle.symbol(), Timestamp.from(cycle.cycleTime()),
                positionState.isFlat() ? "FLAT" : "POSITION",
                riskDecision.allowAction() ? "ACTION_ALLOWED" : "ACTION_BLOCKED",
                riskDecision.blocker());
    }

    void insertModelOutput(TraderModelOutput modelOutput, TraderMarketSnapshot snapshot) {
        jdbcTemplate.update("""
                        insert into trader_model_output
                          (run_id, cycle_id, model_output_id, model_bundle_version, calibration_bundle_version,
                           direction_json, entry_json, continue_json, exit_json, risk_json,
                           uncertainty, ood_score, usable, blocker)
                        values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                        """,
                modelOutput.runId(), modelOutput.cycleId(), modelOutput.modelOutputId(),
                modelOutput.modelBundleVersion(), modelOutput.calibrationBundleVersion(),
                jsonCodec.toJson(modelOutput.direction()), jsonCodec.toJson(modelOutput.entry()),
                jsonCodec.toJson(modelOutput.continuation()), jsonCodec.toJson(modelOutput.exit()),
                jsonCodec.toJson(modelOutput.risk()), modelOutput.uncertainty(), modelOutput.oodScore(),
                snapshot.dataReady(), snapshot.dataReady() ? null : "DATA_NOT_READY");
    }

    void insertPmDecision(TraderPositionManagerDecision decision) {
        jdbcTemplate.update("""
                        insert into trader_position_manager_decision
                          (run_id, cycle_id, pm_decision_id, model_output_id, position_state_id, account_state_id,
                           execution_state_id, candidate_action, side, price_plan_id, price_plan_config_hash,
                           target_position_ratio, add_ratio, reduce_ratio, stop_price, target_price, reason, decision_json)
                        values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                        """,
                decision.runId(), decision.cycleId(), decision.pmDecisionId(), decision.modelOutputId(),
                decision.positionStateId(), decision.accountStateId(), decision.executionStateId(),
                decision.candidateAction().name(), decision.side().name(), decision.pricePlanId(),
                decision.pricePlanConfigHash(), decision.targetPositionRatio(), decision.addRatio(),
                decision.reduceRatio(), decision.stopPrice(), decision.targetPrice(), decision.reason(),
                jsonCodec.toJson(decision.decisionJson()));
    }

    void insertRiskDecision(TraderRiskDecision decision) {
        jdbcTemplate.update("""
                        insert into trader_risk_decision
                          (run_id, cycle_id, risk_decision_id, pm_decision_id, original_action, final_action,
                           allow_action, blocker, decision_json)
                        values (?, ?, ?, ?, ?, ?, ?, ?, ?)
                        """,
                decision.runId(), decision.cycleId(), decision.riskDecisionId(), decision.pmDecisionId(),
                decision.originalAction().name(), decision.finalAction().name(), decision.allowAction(),
                decision.blocker(), jsonCodec.toJson(decision.decisionJson()));
    }

    void insertAction(TraderAction action) {
        jdbcTemplate.update("""
                        insert into trader_action
                          (run_id, cycle_id, action_id, model_output_id, pm_decision_id, risk_decision_id,
                           action_type, symbol, side, price_plan_id, price_plan_config_hash, position_ratio,
                           quantity, stop_price, target_price, reduce_only, idempotency_key, send_status,
                           reason, action_context_json)
                        values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                        """,
                action.runId(), action.cycleId(), action.actionId(), action.modelOutputId(),
                action.pmDecisionId(), action.riskDecisionId(), action.actionType().name(),
                action.symbol(), action.side().name(), action.pricePlanId(), action.pricePlanConfigHash(),
                action.positionRatio(), action.quantity(), action.stopPrice(), action.targetPrice(),
                action.reduceOnly(), action.idempotencyKey(), "PENDING", action.reason(),
                jsonCodec.toJson(action.actionContextJson()));
    }
}