From b5e34b6d186a485386e4092e58db00a3da78b68e Mon Sep 17 00:00:00 2001
From: Lee <1633292@qq.com>
Date: Fri, 24 Jul 2026 12:56:20 +0800
Subject: [PATCH] Add persisted technical indicator calculation

---
 .../backtestify/entity/KLineIndicator.java    |  49 ++++
 .../backtestify/entity/KLineRecord.java       |   2 +-
 .../backtestify/mapper/IndicatorMapper.java   |  77 +++++
 .../backtestify/mapper/KLineMapper.java       |  46 ++-
 .../market/MysqlMarketDataProvider.java       |   2 +-
 .../IndicatorCalculationService.java          |  21 ++
 .../market/indicator/IndicatorCalculator.java |  19 ++
 .../TechnicalIndicatorCalculator.java         | 275 ++++++++++++++++++
 .../impl/IndicatorCalculationServiceImpl.java |  92 ++++++
 .../market/repository/KLineTableResolver.java |   9 +-
 .../market/sync/MarketDataSyncService.java    |   5 +-
 .../IndicatorCalculationServiceImplTest.java  | 125 ++++++++
 .../IndicatorPersistenceIntegrationTest.java  |  82 ++++++
 .../TechnicalIndicatorCalculatorTest.java     | 159 ++++++++++
 .../repository/KLineTableResolverTest.java    |  10 +-
 .../sync/MarketDataSyncServiceTest.java       |  85 ++++++
 src/test/resources/schema-test.sql            |  64 +++-
 17 files changed, 1095 insertions(+), 27 deletions(-)
 create mode 100644 src/main/java/com/yangwale/backtestify/entity/KLineIndicator.java
 create mode 100644 src/main/java/com/yangwale/backtestify/mapper/IndicatorMapper.java
 create mode 100644 src/main/java/com/yangwale/backtestify/service/market/indicator/IndicatorCalculationService.java
 create mode 100644 src/main/java/com/yangwale/backtestify/service/market/indicator/IndicatorCalculator.java
 create mode 100644 src/main/java/com/yangwale/backtestify/service/market/indicator/TechnicalIndicatorCalculator.java
 create mode 100644 src/main/java/com/yangwale/backtestify/service/market/indicator/impl/IndicatorCalculationServiceImpl.java
 create mode 100644 src/test/java/com/yangwale/backtestify/service/market/indicator/IndicatorCalculationServiceImplTest.java
 create mode 100644 src/test/java/com/yangwale/backtestify/service/market/indicator/IndicatorPersistenceIntegrationTest.java
 create mode 100644 src/test/java/com/yangwale/backtestify/service/market/indicator/TechnicalIndicatorCalculatorTest.java
 create mode 100644 src/test/java/com/yangwale/backtestify/service/market/sync/MarketDataSyncServiceTest.java

diff --git a/src/main/java/com/yangwale/backtestify/entity/KLineIndicator.java b/src/main/java/com/yangwale/backtestify/entity/KLineIndicator.java
new file mode 100644
index 0000000..041160f
--- /dev/null
+++ b/src/main/java/com/yangwale/backtestify/entity/KLineIndicator.java
@@ -0,0 +1,49 @@
+package com.yangwale.backtestify.entity;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import java.math.BigDecimal;
+
+/**
+ * K线指标记录，映射各周期同结构的 t_indicator_* 表。
+ */
+@Getter
+@Setter
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class KLineIndicator {
+
+    private Integer instrumentId;
+    private Long kTime;
+
+    private BigDecimal ma5;
+    private BigDecimal ma10;
+    private BigDecimal ma20;
+    private BigDecimal ma60;
+
+    private BigDecimal bollMb;
+    private BigDecimal bollUp;
+    private BigDecimal bollDn;
+
+    private BigDecimal ema6;
+    private BigDecimal ema12;
+    private BigDecimal ema20;
+
+    private BigDecimal macdDif;
+    private BigDecimal macdDea;
+    private BigDecimal macdBar;
+
+    private BigDecimal rsi6;
+    private BigDecimal rsi12;
+
+    private BigDecimal kdjK;
+    private BigDecimal kdjD;
+    private BigDecimal kdjJ;
+
+    private Long createAt;
+}
diff --git a/src/main/java/com/yangwale/backtestify/entity/KLineRecord.java b/src/main/java/com/yangwale/backtestify/entity/KLineRecord.java
index 52a81e3..88212ee 100644
--- a/src/main/java/com/yangwale/backtestify/entity/KLineRecord.java
+++ b/src/main/java/com/yangwale/backtestify/entity/KLineRecord.java
@@ -15,7 +15,7 @@ public class KLineRecord {
     private Integer instrumentId;
 
     /** Unix时间戳，秒级 */
-    private Long timestamp;
+    private Long kTime;
 
     private Integer open;
 
diff --git a/src/main/java/com/yangwale/backtestify/mapper/IndicatorMapper.java b/src/main/java/com/yangwale/backtestify/mapper/IndicatorMapper.java
new file mode 100644
index 0000000..96383fa
--- /dev/null
+++ b/src/main/java/com/yangwale/backtestify/mapper/IndicatorMapper.java
@@ -0,0 +1,77 @@
+package com.yangwale.backtestify.mapper;
+
+import com.yangwale.backtestify.entity.KLineIndicator;
+import org.apache.ibatis.annotations.Insert;
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+import org.apache.ibatis.annotations.Select;
+
+import java.util.List;
+
+@Mapper
+public interface IndicatorMapper {
+
+    @Select("""
+            SELECT instrument_id, k_time,
+                   ma5, ma10, ma20, ma60,
+                   boll_mb, boll_up, boll_dn,
+                   ema6, ema12, ema20,
+                   macd_dif, macd_dea, macd_bar,
+                   rsi6, rsi12,
+                   kdj_k, kdj_d, kdj_j,
+                   create_at
+            FROM ${tableName}
+            WHERE instrument_id = #{instrumentId}
+            ORDER BY k_time DESC
+            LIMIT 1
+            """)
+    KLineIndicator selectLatest(@Param("tableName") String tableName,
+                                @Param("instrumentId") Integer instrumentId);
+
+    @Insert("""
+            <script>
+            INSERT INTO ${tableName}
+                (instrument_id, k_time,
+                 ma5, ma10, ma20, ma60,
+                 boll_mb, boll_up, boll_dn,
+                 ema6, ema12, ema20,
+                 macd_dif, macd_dea, macd_bar,
+                 rsi6, rsi12,
+                 kdj_k, kdj_d, kdj_j,
+                 create_at)
+            VALUES
+            <foreach collection="records" item="item" separator=",">
+                (#{item.instrumentId}, #{item.kTime},
+                 #{item.ma5}, #{item.ma10}, #{item.ma20}, #{item.ma60},
+                 #{item.bollMb}, #{item.bollUp}, #{item.bollDn},
+                 #{item.ema6}, #{item.ema12}, #{item.ema20},
+                 #{item.macdDif}, #{item.macdDea}, #{item.macdBar},
+                 #{item.rsi6}, #{item.rsi12},
+                 #{item.kdjK}, #{item.kdjD}, #{item.kdjJ},
+                 #{item.createAt})
+            </foreach>
+            ON DUPLICATE KEY UPDATE
+                ma5 = VALUES(ma5),
+                ma10 = VALUES(ma10),
+                ma20 = VALUES(ma20),
+                ma60 = VALUES(ma60),
+                boll_mb = VALUES(boll_mb),
+                boll_up = VALUES(boll_up),
+                boll_dn = VALUES(boll_dn),
+                ema6 = VALUES(ema6),
+                ema12 = VALUES(ema12),
+                ema20 = VALUES(ema20),
+                macd_dif = VALUES(macd_dif),
+                macd_dea = VALUES(macd_dea),
+                macd_bar = VALUES(macd_bar),
+                rsi6 = VALUES(rsi6),
+                rsi12 = VALUES(rsi12),
+                kdj_k = VALUES(kdj_k),
+                kdj_d = VALUES(kdj_d),
+                kdj_j = VALUES(kdj_j),
+                create_at = VALUES(create_at)
+            </script>
+            """)
+    int upsertBatch(@Param("tableName") String tableName,
+                    @Param("records") List<KLineIndicator> records);
+}
diff --git a/src/main/java/com/yangwale/backtestify/mapper/KLineMapper.java b/src/main/java/com/yangwale/backtestify/mapper/KLineMapper.java
index 71fd18f..cc941cd 100644
--- a/src/main/java/com/yangwale/backtestify/mapper/KLineMapper.java
+++ b/src/main/java/com/yangwale/backtestify/mapper/KLineMapper.java
@@ -12,25 +12,61 @@ import java.util.List;
 public interface KLineMapper {
 
     @Select("""
-            SELECT instrument_id, timestamp, open, high, low, close,
+            SELECT instrument_id, k_time, open, high, low, close,
                    volume, turnover, open_interest
             FROM ${tableName}
             WHERE instrument_id = #{instrumentId}
-              AND timestamp BETWEEN #{startTimestamp} AND #{endTimestamp}
-            ORDER BY timestamp ASC
+              AND k_time BETWEEN #{startTimestamp} AND #{endTimestamp}
+            ORDER BY k_time ASC
             """)
     List<KLineRecord> selectRange(@Param("tableName") String tableName,
                                   @Param("instrumentId") Integer instrumentId,
                                   @Param("startTimestamp") long startTimestamp,
                                   @Param("endTimestamp") long endTimestamp);
 
+    @Select("""
+            SELECT instrument_id, k_time, open, high, low, close,
+                   volume, turnover, open_interest
+            FROM ${tableName}
+            WHERE instrument_id = #{instrumentId}
+            ORDER BY k_time ASC
+            """)
+    List<KLineRecord> selectAll(@Param("tableName") String tableName,
+                                @Param("instrumentId") Integer instrumentId);
+
+    @Select("""
+            SELECT instrument_id, k_time, open, high, low, close,
+                   volume, turnover, open_interest
+            FROM ${tableName}
+            WHERE instrument_id = #{instrumentId}
+              AND k_time > #{afterTimestamp}
+            ORDER BY k_time ASC
+            """)
+    List<KLineRecord> selectAfter(@Param("tableName") String tableName,
+                                  @Param("instrumentId") Integer instrumentId,
+                                  @Param("afterTimestamp") long afterTimestamp);
+
+    @Select("""
+            SELECT instrument_id, k_time, open, high, low, close,
+                   volume, turnover, open_interest
+            FROM ${tableName}
+            WHERE instrument_id = #{instrumentId}
+              AND k_time < #{beforeTimestamp}
+            ORDER BY k_time DESC
+            LIMIT #{limit}
+            """)
+    List<KLineRecord> selectBefore(@Param("tableName") String tableName,
+                                   @Param("instrumentId") Integer instrumentId,
+                                   @Param("beforeTimestamp") long beforeTimestamp,
+                                   @Param("limit") int limit);
+
     @Insert("""
             <script>
             INSERT INTO ${tableName}
-                (instrument_id, timestamp, open, high, low, close, volume, turnover, open_interest)
+                (instrument_id, k_time, open, high, low, close, volume, turnover, open_interest)
             VALUES
             <foreach collection="records" item="item" separator=",">
-                (#{item.instrumentId}, #{item.timestamp}, #{item.open}, #{item.high}, #{item.low}, #{item.close},
+                (#{item.instrumentId}, #{item.kTime}, #{item.open}, #{item.high}, #{item.low}, #{item.close},
                  #{item.volume}, #{item.turnover}, #{item.openInterest})
             </foreach>
             ON DUPLICATE KEY UPDATE
diff --git a/src/main/java/com/yangwale/backtestify/service/market/MysqlMarketDataProvider.java b/src/main/java/com/yangwale/backtestify/service/market/MysqlMarketDataProvider.java
index 31769b6..2ac8d87 100644
--- a/src/main/java/com/yangwale/backtestify/service/market/MysqlMarketDataProvider.java
+++ b/src/main/java/com/yangwale/backtestify/service/market/MysqlMarketDataProvider.java
@@ -84,7 +84,7 @@ public class MysqlMarketDataProvider implements MarketDataProvider {
 
     private KLineData toKLineData(KLineRecord record, int priceScale) {
         return KLineData.builder()
-                .time(LocalDateTime.ofInstant(Instant.ofEpochSecond(record.getTimestamp()), ZONE_ID))
+                .time(LocalDateTime.ofInstant(Instant.ofEpochSecond(record.getKTime()), ZONE_ID))
                 .open(PriceScaleConverter.toRaw(record.getOpen(), priceScale))
                 .high(PriceScaleConverter.toRaw(record.getHigh(), priceScale))
                 .low(PriceScaleConverter.toRaw(record.getLow(), priceScale))
diff --git a/src/main/java/com/yangwale/backtestify/service/market/indicator/IndicatorCalculationService.java b/src/main/java/com/yangwale/backtestify/service/market/indicator/IndicatorCalculationService.java
new file mode 100644
index 0000000..5ab423e
--- /dev/null
+++ b/src/main/java/com/yangwale/backtestify/service/market/indicator/IndicatorCalculationService.java
@@ -0,0 +1,21 @@
+package com.yangwale.backtestify.service.market.indicator;
+
+/**
+ * 指标历史重建及增量更新入口。
+ */
+public interface IndicatorCalculationService {
+
+    /**
+     * 计算指定合约、周期的全部历史K线指标并批量写入。
+     *
+     * @return 数据库受影响行数
+     */
+    int rebuild(String period, Integer instrumentId);
+
+    /**
+     * 从指标表最新时间点之后计算新增K线，只写入新增指标。
+     *
+     * @return 数据库受影响行数
+     */
+    int updateIncremental(String period, Integer instrumentId);
+}
diff --git a/src/main/java/com/yangwale/backtestify/service/market/indicator/IndicatorCalculator.java b/src/main/java/com/yangwale/backtestify/service/market/indicator/IndicatorCalculator.java
new file mode 100644
index 0000000..d375c0a
--- /dev/null
+++ b/src/main/java/com/yangwale/backtestify/service/market/indicator/IndicatorCalculator.java
@@ -0,0 +1,19 @@
+package com.yangwale.backtestify.service.market.indicator;
+
+import com.yangwale.backtestify.entity.KLineIndicator;
+import com.yangwale.backtestify.entity.KLineRecord;
+
+import java.util.List;
+
+/**
+ * K线技术指标计算边界。
+ */
+public interface IndicatorCalculator {
+
+    List<KLineIndicator> calculate(List<KLineRecord> kLines, int priceScale);
+
+    List<KLineIndicator> calculateIncremental(List<KLineRecord> history,
+                                              List<KLineRecord> additions,
+                                              KLineIndicator previousIndicator,
+                                              int priceScale);
+}
diff --git a/src/main/java/com/yangwale/backtestify/service/market/indicator/TechnicalIndicatorCalculator.java b/src/main/java/com/yangwale/backtestify/service/market/indicator/TechnicalIndicatorCalculator.java
new file mode 100644
index 0000000..7ac51fb
--- /dev/null
+++ b/src/main/java/com/yangwale/backtestify/service/market/indicator/TechnicalIndicatorCalculator.java
@@ -0,0 +1,275 @@
+package com.yangwale.backtestify.service.market.indicator;
+
+import com.yangwale.backtestify.entity.KLineIndicator;
+import com.yangwale.backtestify.entity.KLineRecord;
+import org.springframework.stereotype.Component;
+
+import java.math.BigDecimal;
+import java.math.MathContext;
+import java.math.RoundingMode;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+
+/**
+ * MA、BOLL、EMA、MACD、RSI、KDJ 的统一计算器。
+ *
+ * <p>价格类指标保留 6 位小数，摆动类指标保留 4 位小数，与指标表字段精度一致。</p>
+ */
+@Component
+public class TechnicalIndicatorCalculator implements IndicatorCalculator {
+
+    private static final MathContext MC = MathContext.DECIMAL128;
+    private static final BigDecimal TWO = BigDecimal.valueOf(2);
+    private static final BigDecimal FIFTY = BigDecimal.valueOf(50);
+    private static final BigDecimal HUNDRED = BigDecimal.valueOf(100);
+
+    @Override
+    public List<KLineIndicator> calculate(List<KLineRecord> kLines, int priceScale) {
+        validate(kLines, priceScale);
+        return calculateRows(List.of(), kLines, new State(), priceScale);
+    }
+
+    @Override
+    public List<KLineIndicator> calculateIncremental(List<KLineRecord> history,
+                                                     List<KLineRecord> additions,
+                                                     KLineIndicator previousIndicator,
+                                                     int priceScale) {
+        validate(history, priceScale);
+        validate(additions, priceScale);
+        if (additions.isEmpty()) {
+            return List.of();
+        }
+        if (previousIndicator == null) {
+            List<KLineRecord> all = new ArrayList<>(history.size() + additions.size());
+            all.addAll(history);
+            all.addAll(additions);
+            List<KLineIndicator> calculated = calculate(all, priceScale);
+            return List.copyOf(calculated.subList(calculated.size() - additions.size(), calculated.size()));
+        }
+
+        State state = State.from(previousIndicator);
+        return calculateRows(history, additions, state, priceScale);
+    }
+
+    private List<KLineIndicator> calculateRows(List<KLineRecord> history,
+                                                List<KLineRecord> rows,
+                                                State state,
+                                                int priceScale) {
+        List<PriceBar> window = new ArrayList<>(history.size() + rows.size());
+        for (KLineRecord record : history) {
+            window.add(toPriceBar(record, priceScale));
+        }
+
+        List<KLineIndicator> result = new ArrayList<>(rows.size());
+        long createAt = Instant.now().getEpochSecond();
+        for (KLineRecord record : rows) {
+            PriceBar current = toPriceBar(record, priceScale);
+            window.add(current);
+
+            BigDecimal ma5 = movingAverage(window, 5);
+            BigDecimal ma10 = movingAverage(window, 10);
+            BigDecimal ma20 = movingAverage(window, 20);
+            BigDecimal ma60 = movingAverage(window, 60);
+
+            BigDecimal bollMb = null;
+            BigDecimal bollUp = null;
+            BigDecimal bollDn = null;
+            if (window.size() >= 20) {
+                bollMb = price(movingAverageRaw(window, 20));
+                BigDecimal standardDeviation = standardDeviation(window, 20);
+                bollUp = price(bollMb.add(standardDeviation.multiply(TWO, MC), MC));
+                bollDn = price(bollMb.subtract(standardDeviation.multiply(TWO, MC), MC));
+            }
+
+            state.ema6 = ema(current.close(), state.ema6, 6);
+            state.ema12 = ema(current.close(), state.ema12, 12);
+            state.ema20 = ema(current.close(), state.ema20, 20);
+            state.ema26 = ema(current.close(), state.ema26, 26);
+
+            BigDecimal dif = state.ema12.subtract(state.ema26, MC);
+            state.dea = state.dea == null
+                    ? dif
+                    : state.dea.multiply(BigDecimal.valueOf(0.8), MC)
+                    .add(dif.multiply(BigDecimal.valueOf(0.2), MC), MC);
+            BigDecimal macdBar = price(dif.subtract(state.dea, MC).multiply(TWO, MC));
+
+            BigDecimal rsi6 = rsi(window, 6);
+            BigDecimal rsi12 = rsi(window, 12);
+
+            BigDecimal rsv = rsv(window, 9);
+            state.k = state.k.multiply(BigDecimal.valueOf(2), MC)
+                    .add(rsv, MC).divide(BigDecimal.valueOf(3), MC);
+            state.d = state.d.multiply(BigDecimal.valueOf(2), MC)
+                    .add(state.k, MC).divide(BigDecimal.valueOf(3), MC);
+            BigDecimal j = oscillator(state.k.multiply(BigDecimal.valueOf(3), MC)
+                    .subtract(state.d.multiply(BigDecimal.valueOf(2), MC), MC));
+
+            result.add(KLineIndicator.builder()
+                    .instrumentId(record.getInstrumentId())
+                    .kTime(record.getKTime())
+                    .ma5(ma5)
+                    .ma10(ma10)
+                    .ma20(ma20)
+                    .ma60(ma60)
+                    .bollMb(bollMb)
+                    .bollUp(bollUp)
+                    .bollDn(bollDn)
+                    .ema6(price(state.ema6))
+                    .ema12(price(state.ema12))
+                    .ema20(price(state.ema20))
+                    .macdDif(price(dif))
+                    .macdDea(price(state.dea))
+                    .macdBar(macdBar)
+                    .rsi6(rsi6)
+                    .rsi12(rsi12)
+                    .kdjK(oscillator(state.k))
+                    .kdjD(oscillator(state.d))
+                    .kdjJ(j)
+                    .createAt(createAt)
+                    .build());
+        }
+        return result;
+    }
+
+    private BigDecimal movingAverage(List<PriceBar> values, int period) {
+        return values.size() < period ? null : price(movingAverageRaw(values, period));
+    }
+
+    private BigDecimal movingAverageRaw(List<PriceBar> values, int period) {
+        BigDecimal sum = BigDecimal.ZERO;
+        for (int i = values.size() - period; i < values.size(); i++) {
+            sum = sum.add(values.get(i).close(), MC);
+        }
+        return sum.divide(BigDecimal.valueOf(period), MC);
+    }
+
+    private BigDecimal standardDeviation(List<PriceBar> values, int period) {
+        BigDecimal mean = movingAverageRaw(values, period);
+        BigDecimal squareSum = BigDecimal.ZERO;
+        for (int i = values.size() - period; i < values.size(); i++) {
+            BigDecimal difference = values.get(i).close().subtract(mean, MC);
+            squareSum = squareSum.add(difference.multiply(difference, MC), MC);
+        }
+        return squareSum.divide(BigDecimal.valueOf(period), MC).sqrt(MC);
+    }
+
+    private BigDecimal ema(BigDecimal close, BigDecimal previous, int period) {
+        if (previous == null) {
+            return close;
+        }
+        BigDecimal multiplier = TWO.divide(BigDecimal.valueOf(period + 1L), MC);
+        return close.subtract(previous, MC).multiply(multiplier, MC).add(previous, MC);
+    }
+
+    private BigDecimal rsi(List<PriceBar> values, int period) {
+        if (values.size() <= period) {
+            return null;
+        }
+        BigDecimal gains = BigDecimal.ZERO;
+        BigDecimal losses = BigDecimal.ZERO;
+        int start = values.size() - period;
+        for (int i = start; i < values.size(); i++) {
+            BigDecimal change = values.get(i).close().subtract(values.get(i - 1).close(), MC);
+            if (change.signum() > 0) {
+                gains = gains.add(change, MC);
+            } else {
+                losses = losses.add(change.abs(), MC);
+            }
+        }
+        if (gains.signum() == 0 && losses.signum() == 0) {
+            return oscillator(FIFTY);
+        }
+        if (losses.signum() == 0) {
+            return oscillator(HUNDRED);
+        }
+        return oscillator(gains.divide(gains.add(losses, MC), MC).multiply(HUNDRED, MC));
+    }
+
+    private BigDecimal rsv(List<PriceBar> values, int period) {
+        int start = Math.max(0, values.size() - period);
+        BigDecimal highest = values.get(start).high();
+        BigDecimal lowest = values.get(start).low();
+        for (int i = start + 1; i < values.size(); i++) {
+            highest = highest.max(values.get(i).high());
+            lowest = lowest.min(values.get(i).low());
+        }
+        BigDecimal range = highest.subtract(lowest, MC);
+        if (range.signum() == 0) {
+            return FIFTY;
+        }
+        return values.getLast().close().subtract(lowest, MC)
+                .divide(range, MC)
+                .multiply(HUNDRED, MC);
+    }
+
+    private PriceBar toPriceBar(KLineRecord record, int priceScale) {
+        BigDecimal divisor = BigDecimal.valueOf(priceScale);
+        return new PriceBar(
+                BigDecimal.valueOf(record.getHigh()).divide(divisor, MC),
+                BigDecimal.valueOf(record.getLow()).divide(divisor, MC),
+                BigDecimal.valueOf(record.getClose()).divide(divisor, MC));
+    }
+
+    private BigDecimal price(BigDecimal value) {
+        return value.setScale(6, RoundingMode.HALF_UP);
+    }
+
+    private BigDecimal oscillator(BigDecimal value) {
+        return value.setScale(4, RoundingMode.HALF_UP);
+    }
+
+    private void validate(List<KLineRecord> records, int priceScale) {
+        if (records == null) {
+            throw new IllegalArgumentException("K线列表不能为空");
+        }
+        if (priceScale <= 0) {
+            throw new IllegalArgumentException("价格缩放倍数必须大于0");
+        }
+        if (records.stream().anyMatch(record -> record == null
+                || record.getKTime() == null
+                || record.getHigh() == null
+                || record.getLow() == null
+                || record.getClose() == null)) {
+            throw new IllegalArgumentException("K线时间及高低收价格不能为空");
+        }
+        if (!records.equals(records.stream()
+                .sorted(Comparator.comparing(KLineRecord::getKTime))
+                .toList())) {
+            throw new IllegalArgumentException("K线必须按时间升序排列");
+        }
+    }
+
+    private record PriceBar(BigDecimal high, BigDecimal low, BigDecimal close) {
+    }
+
+    private static final class State {
+        private BigDecimal ema6;
+        private BigDecimal ema12;
+        private BigDecimal ema20;
+        private BigDecimal ema26;
+        private BigDecimal dea;
+        private BigDecimal k = oscillatorValue(FIFTY);
+        private BigDecimal d = oscillatorValue(FIFTY);
+
+        private static State from(KLineIndicator previous) {
+            State state = new State();
+            state.ema6 = previous.getEma6();
+            state.ema12 = previous.getEma12();
+            state.ema20 = previous.getEma20();
+            state.ema26 = previous.getEma12().subtract(previous.getMacdDif(), MC)
+                    .setScale(6, RoundingMode.HALF_UP);
+            state.dea = previous.getMacdDea() != null
+                    ? previous.getMacdDea()
+                    : previous.getMacdDif();
+            state.k = previous.getKdjK();
+            state.d = previous.getKdjD();
+            return state;
+        }
+
+        private static BigDecimal oscillatorValue(BigDecimal value) {
+            return value.setScale(4, RoundingMode.HALF_UP);
+        }
+    }
+}
diff --git a/src/main/java/com/yangwale/backtestify/service/market/indicator/impl/IndicatorCalculationServiceImpl.java b/src/main/java/com/yangwale/backtestify/service/market/indicator/impl/IndicatorCalculationServiceImpl.java
new file mode 100644
index 0000000..8afbfa5
--- /dev/null
+++ b/src/main/java/com/yangwale/backtestify/service/market/indicator/impl/IndicatorCalculationServiceImpl.java
@@ -0,0 +1,92 @@
+package com.yangwale.backtestify.service.market.indicator.impl;
+
+import com.yangwale.backtestify.common.BusinessException;
+import com.yangwale.backtestify.common.ErrorCode;
+import com.yangwale.backtestify.entity.InstrumentDictionary;
+import com.yangwale.backtestify.entity.KLineIndicator;
+import com.yangwale.backtestify.entity.KLineRecord;
+import com.yangwale.backtestify.mapper.IndicatorMapper;
+import com.yangwale.backtestify.mapper.InstrumentDictionaryMapper;
+import com.yangwale.backtestify.mapper.KLineMapper;
+import com.yangwale.backtestify.service.market.indicator.IndicatorCalculationService;
+import com.yangwale.backtestify.service.market.indicator.IndicatorCalculator;
+import com.yangwale.backtestify.service.market.repository.KLineTableResolver;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+@Service
+@RequiredArgsConstructor
+public class IndicatorCalculationServiceImpl implements IndicatorCalculationService {
+
+    private static final int CONTEXT_SIZE = 60;
+    private static final int WRITE_BATCH_SIZE = 500;
+
+    private final InstrumentDictionaryMapper instrumentMapper;
+    private final KLineMapper kLineMapper;
+    private final IndicatorMapper indicatorMapper;
+    private final KLineTableResolver tableResolver;
+    private final IndicatorCalculator calculator;
+
+    @Override
+    @Transactional(rollbackFor = Exception.class)
+    public int rebuild(String period, Integer instrumentId) {
+        InstrumentDictionary instrument = requireInstrument(instrumentId);
+        String kLineTable = tableResolver.resolve(period);
+        String indicatorTable = tableResolver.resolveIndicator(period);
+        List<KLineRecord> kLines = kLineMapper.selectAll(kLineTable, instrumentId);
+        if (kLines.isEmpty()) {
+            return 0;
+        }
+        List<KLineIndicator> indicators = calculator.calculate(kLines, instrument.getPriceScale());
+        return upsertBatches(indicatorTable, indicators);
+    }
+
+    @Override
+    @Transactional(rollbackFor = Exception.class)
+    public int updateIncremental(String period, Integer instrumentId) {
+        InstrumentDictionary instrument = requireInstrument(instrumentId);
+        String kLineTable = tableResolver.resolve(period);
+        String indicatorTable = tableResolver.resolveIndicator(period);
+        KLineIndicator previous = indicatorMapper.selectLatest(indicatorTable, instrumentId);
+        if (previous == null) {
+            return rebuild(period, instrumentId);
+        }
+
+        List<KLineRecord> additions = kLineMapper.selectAfter(
+                kLineTable, instrumentId, previous.getKTime());
+        if (additions.isEmpty()) {
+            return 0;
+        }
+
+        List<KLineRecord> history = new ArrayList<>(kLineMapper.selectBefore(
+                kLineTable, instrumentId, additions.getFirst().getKTime(), CONTEXT_SIZE));
+        Collections.reverse(history);
+        List<KLineIndicator> indicators = calculator.calculateIncremental(
+                history, additions, previous, instrument.getPriceScale());
+        return upsertBatches(indicatorTable, indicators);
+    }
+
+    private InstrumentDictionary requireInstrument(Integer instrumentId) {
+        InstrumentDictionary instrument = instrumentMapper.selectById(instrumentId);
+        if (instrument == null) {
+            throw new BusinessException(ErrorCode.CONTRACT_NOT_FOUND,
+                    "合约不存在, instrumentId=" + instrumentId);
+        }
+        return instrument;
+    }
+
+    private int upsertBatches(String tableName, List<KLineIndicator> indicators) {
+        int affectedRows = 0;
+        for (int start = 0; start < indicators.size(); start += WRITE_BATCH_SIZE) {
+            int end = Math.min(start + WRITE_BATCH_SIZE, indicators.size());
+            affectedRows += indicatorMapper.upsertBatch(
+                    tableName, List.copyOf(indicators.subList(start, end)));
+        }
+        return affectedRows;
+    }
+}
diff --git a/src/main/java/com/yangwale/backtestify/service/market/repository/KLineTableResolver.java b/src/main/java/com/yangwale/backtestify/service/market/repository/KLineTableResolver.java
index 1c101d8..938cf63 100644
--- a/src/main/java/com/yangwale/backtestify/service/market/repository/KLineTableResolver.java
+++ b/src/main/java/com/yangwale/backtestify/service/market/repository/KLineTableResolver.java
@@ -15,13 +15,15 @@ public class KLineTableResolver {
 
     private static final Map<String, String> TABLES = Map.of(
             "1m", "t_kline_1m",
+            "3m", "t_kline_3m",
             "5m", "t_kline_5m",
             "15m", "t_kline_15m",
             "30m", "t_kline_30m",
             "1h", "t_kline_1h",
             "4h", "t_kline_4h",
             "1d", "t_kline_1d",
-            "1w", "t_kline_1w"
+            "1w", "t_kline_1w",
+            "1mo", "t_kline_1mo"
     );
 
     public String resolve(String period) {
@@ -33,6 +35,11 @@ public class KLineTableResolver {
         return tableName;
     }
 
+    public String resolveIndicator(String period) {
+        String kLineTable = resolve(period);
+        return kLineTable.replace("t_kline_", "t_indicator_");
+    }
+
     public Set<String> supportedPeriods() {
         return TABLES.keySet();
     }
diff --git a/src/main/java/com/yangwale/backtestify/service/market/sync/MarketDataSyncService.java b/src/main/java/com/yangwale/backtestify/service/market/sync/MarketDataSyncService.java
index b5ce0e0..4412f67 100644
--- a/src/main/java/com/yangwale/backtestify/service/market/sync/MarketDataSyncService.java
+++ b/src/main/java/com/yangwale/backtestify/service/market/sync/MarketDataSyncService.java
@@ -12,6 +12,7 @@ import com.yangwale.backtestify.mapper.MarketDataSyncLogMapper;
 import com.yangwale.backtestify.service.market.client.CnQuotationClient;
 import com.yangwale.backtestify.service.market.client.CnQuotationModels;
 import com.yangwale.backtestify.service.market.convert.PriceScaleConverter;
+import com.yangwale.backtestify.service.market.indicator.IndicatorCalculationService;
 import com.yangwale.backtestify.service.market.repository.KLineTableResolver;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -41,6 +42,7 @@ public class MarketDataSyncService {
     private final KLineMapper kLineMapper;
     private final MarketDataSyncLogMapper syncLogMapper;
     private final KLineTableResolver tableResolver;
+    private final IndicatorCalculationService indicatorCalculationService;
 
     @Scheduled(cron = "${market-data.sync.cron:0 0 6 * * ?}", zone = "${market-data.sync.zone:Asia/Shanghai}")
     public void syncYesterdayMainContracts() {
@@ -111,6 +113,7 @@ public class MarketDataSyncService {
                     .toList();
             if (!records.isEmpty()) {
                 count = kLineMapper.upsertBatch(tableResolver.resolve(period), records);
+                indicatorCalculationService.updateIncremental(period, instrument.getId());
             }
             saveLog("KLINE", period, instrument.getContractCode(), syncDate, "SUCCESS", count, null, start);
         } catch (Exception e) {
@@ -148,7 +151,7 @@ public class MarketDataSyncService {
         try {
             return KLineRecord.builder()
                     .instrumentId(instrument.getId())
-                    .timestamp(item.u())
+                    .kTime(item.u())
                     .open(PriceScaleConverter.toScaled(item.o(), instrument.getPriceScale()))
                     .high(PriceScaleConverter.toScaled(item.h(), instrument.getPriceScale()))
                     .low(PriceScaleConverter.toScaled(item.l(), instrument.getPriceScale()))
diff --git a/src/test/java/com/yangwale/backtestify/service/market/indicator/IndicatorCalculationServiceImplTest.java b/src/test/java/com/yangwale/backtestify/service/market/indicator/IndicatorCalculationServiceImplTest.java
new file mode 100644
index 0000000..56ba196
--- /dev/null
+++ b/src/test/java/com/yangwale/backtestify/service/market/indicator/IndicatorCalculationServiceImplTest.java
@@ -0,0 +1,125 @@
+package com.yangwale.backtestify.service.market.indicator;
+
+import com.yangwale.backtestify.entity.InstrumentDictionary;
+import com.yangwale.backtestify.entity.KLineIndicator;
+import com.yangwale.backtestify.entity.KLineRecord;
+import com.yangwale.backtestify.mapper.IndicatorMapper;
+import com.yangwale.backtestify.mapper.InstrumentDictionaryMapper;
+import com.yangwale.backtestify.mapper.KLineMapper;
+import com.yangwale.backtestify.service.market.indicator.impl.IndicatorCalculationServiceImpl;
+import com.yangwale.backtestify.service.market.repository.KLineTableResolver;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class IndicatorCalculationServiceImplTest {
+
+    @Mock
+    private InstrumentDictionaryMapper instrumentMapper;
+    @Mock
+    private KLineMapper kLineMapper;
+    @Mock
+    private IndicatorMapper indicatorMapper;
+    @Mock
+    private IndicatorCalculator calculator;
+
+    private IndicatorCalculationService service;
+
+    @BeforeEach
+    void setUp() {
+        service = new IndicatorCalculationServiceImpl(
+                instrumentMapper,
+                kLineMapper,
+                indicatorMapper,
+                new KLineTableResolver(),
+                calculator);
+    }
+
+    @Test
+    void rebuildCalculatesAndUpsertsAllHistoricalRows() {
+        InstrumentDictionary instrument = instrument(7, 100);
+        List<KLineRecord> kLines = List.of(kLine(1), kLine(2));
+        List<KLineIndicator> indicators = List.of(indicator(1), indicator(2));
+        when(instrumentMapper.selectById(7)).thenReturn(instrument);
+        when(kLineMapper.selectAll("t_kline_1d", 7)).thenReturn(kLines);
+        when(calculator.calculate(kLines, 100)).thenReturn(indicators);
+        when(indicatorMapper.upsertBatch("t_indicator_1d", indicators)).thenReturn(2);
+
+        int count = service.rebuild("1d", 7);
+
+        assertEquals(2, count);
+        verify(indicatorMapper).upsertBatch("t_indicator_1d", indicators);
+    }
+
+    @Test
+    void incrementalLoadsContextButOnlyUpsertsRowsAfterLatestIndicator() {
+        InstrumentDictionary instrument = instrument(7, 100);
+        KLineIndicator previous = indicator(100);
+        List<KLineRecord> additions = List.of(kLine(101), kLine(102));
+        List<KLineRecord> descendingContext = List.of(kLine(100), kLine(99));
+        List<KLineIndicator> calculated = List.of(indicator(101), indicator(102));
+        when(instrumentMapper.selectById(7)).thenReturn(instrument);
+        when(indicatorMapper.selectLatest("t_indicator_1d", 7)).thenReturn(previous);
+        when(kLineMapper.selectAfter("t_kline_1d", 7, 100L)).thenReturn(additions);
+        when(kLineMapper.selectBefore("t_kline_1d", 7, 101L, 60)).thenReturn(descendingContext);
+        when(calculator.calculateIncremental(
+                List.of(descendingContext.get(1), descendingContext.get(0)),
+                additions,
+                previous,
+                100)).thenReturn(calculated);
+        when(indicatorMapper.upsertBatch("t_indicator_1d", calculated)).thenReturn(2);
+
+        int count = service.updateIncremental("1d", 7);
+
+        assertEquals(2, count);
+        verify(indicatorMapper).upsertBatch("t_indicator_1d", calculated);
+    }
+
+    @Test
+    void incrementalDoesNothingWhenNoNewKLinesExist() {
+        when(instrumentMapper.selectById(7)).thenReturn(instrument(7, 100));
+        when(indicatorMapper.selectLatest("t_indicator_1d", 7)).thenReturn(indicator(100));
+        when(kLineMapper.selectAfter("t_kline_1d", 7, 100L)).thenReturn(List.of());
+
+        assertEquals(0, service.updateIncremental("1d", 7));
+
+        verify(calculator, never()).calculateIncremental(anyList(), anyList(), org.mockito.ArgumentMatchers.any(), org.mockito.ArgumentMatchers.anyInt());
+        verify(indicatorMapper, never()).upsertBatch(org.mockito.ArgumentMatchers.anyString(), anyList());
+    }
+
+    private InstrumentDictionary instrument(int id, int priceScale) {
+        InstrumentDictionary instrument = new InstrumentDictionary();
+        instrument.setId(id);
+        instrument.setPriceScale(priceScale);
+        return instrument;
+    }
+
+    private KLineRecord kLine(long time) {
+        return KLineRecord.builder()
+                .instrumentId(7)
+                .kTime(time)
+                .open(100)
+                .high(101)
+                .low(99)
+                .close(100)
+                .build();
+    }
+
+    private KLineIndicator indicator(long time) {
+        return KLineIndicator.builder()
+                .instrumentId(7)
+                .kTime(time)
+                .build();
+    }
+}
diff --git a/src/test/java/com/yangwale/backtestify/service/market/indicator/IndicatorPersistenceIntegrationTest.java b/src/test/java/com/yangwale/backtestify/service/market/indicator/IndicatorPersistenceIntegrationTest.java
new file mode 100644
index 0000000..212b99a
--- /dev/null
+++ b/src/test/java/com/yangwale/backtestify/service/market/indicator/IndicatorPersistenceIntegrationTest.java
@@ -0,0 +1,82 @@
+package com.yangwale.backtestify.service.market.indicator;
+
+import com.yangwale.backtestify.entity.InstrumentDictionary;
+import com.yangwale.backtestify.entity.KLineIndicator;
+import com.yangwale.backtestify.entity.KLineRecord;
+import com.yangwale.backtestify.mapper.IndicatorMapper;
+import com.yangwale.backtestify.mapper.InstrumentDictionaryMapper;
+import com.yangwale.backtestify.mapper.KLineMapper;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@SpringBootTest
+@Transactional
+class IndicatorPersistenceIntegrationTest {
+
+    @Autowired
+    private InstrumentDictionaryMapper instrumentMapper;
+    @Autowired
+    private KLineMapper kLineMapper;
+    @Autowired
+    private IndicatorMapper indicatorMapper;
+    @Autowired
+    private IndicatorCalculationService indicatorService;
+
+    @Test
+    void rebuildAndIncrementalUpdatePersistCalculatedIndicators() {
+        InstrumentDictionary instrument = new InstrumentDictionary();
+        instrument.setExchangeId("TEST");
+        instrument.setSymbol("ti");
+        instrument.setContractCode("ti2601");
+        instrument.setPriceScale(100);
+        instrument.setIsMain(0);
+        instrumentMapper.insert(instrument);
+
+        List<KLineRecord> history = records(instrument.getId(), 1, 20);
+        assertTrue(kLineMapper.upsertBatch("t_kline_1m", history) > 0);
+        assertTrue(indicatorService.rebuild("1m", instrument.getId()) > 0);
+
+        KLineIndicator historicalLatest = indicatorMapper.selectLatest(
+                "t_indicator_1m", instrument.getId());
+        assertEquals(20L, historicalLatest.getKTime());
+        assertEquals(new BigDecimal("10.500000"), historicalLatest.getMa20());
+        assertEquals(new BigDecimal("100.0000"), historicalLatest.getRsi12());
+
+        assertTrue(kLineMapper.upsertBatch(
+                "t_kline_1m", records(instrument.getId(), 21, 22)) > 0);
+        assertTrue(indicatorService.updateIncremental("1m", instrument.getId()) > 0);
+
+        KLineIndicator incrementalLatest = indicatorMapper.selectLatest(
+                "t_indicator_1m", instrument.getId());
+        assertEquals(22L, incrementalLatest.getKTime());
+        assertEquals(new BigDecimal("12.500000"), incrementalLatest.getMa20());
+    }
+
+    private List<KLineRecord> records(int instrumentId, int from, int to) {
+        List<KLineRecord> result = new ArrayList<>();
+        for (int value = from; value <= to; value++) {
+            int scaledClose = value * 100;
+            result.add(KLineRecord.builder()
+                    .instrumentId(instrumentId)
+                    .kTime((long) value)
+                    .open(scaledClose)
+                    .high(scaledClose + 100)
+                    .low(scaledClose - 100)
+                    .close(scaledClose)
+                    .volume(100L)
+                    .turnover(1000L)
+                    .openInterest(50L)
+                    .build());
+        }
+        return result;
+    }
+}
diff --git a/src/test/java/com/yangwale/backtestify/service/market/indicator/TechnicalIndicatorCalculatorTest.java b/src/test/java/com/yangwale/backtestify/service/market/indicator/TechnicalIndicatorCalculatorTest.java
new file mode 100644
index 0000000..0364823
--- /dev/null
+++ b/src/test/java/com/yangwale/backtestify/service/market/indicator/TechnicalIndicatorCalculatorTest.java
@@ -0,0 +1,159 @@
+package com.yangwale.backtestify.service.market.indicator;
+
+import com.yangwale.backtestify.entity.KLineIndicator;
+import com.yangwale.backtestify.entity.KLineRecord;
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TechnicalIndicatorCalculatorTest {
+
+    private final TechnicalIndicatorCalculator calculator = new TechnicalIndicatorCalculator();
+
+    @Test
+    void calculatesMaBollEmaAndMacdUsingConfiguredIndustryPeriods() {
+        List<KLineIndicator> values = calculator.calculate(sequence(1, 60), 1);
+
+        assertNull(values.get(3).getMa5());
+        assertDecimal("3.000000", values.get(4).getMa5());
+        assertDecimal("5.500000", values.get(9).getMa10());
+        assertDecimal("10.500000", values.get(19).getMa20());
+        assertDecimal("30.500000", values.get(59).getMa60());
+
+        assertNull(values.get(18).getBollMb());
+        assertDecimal("10.500000", values.get(19).getBollMb());
+        assertDecimal("22.032563", values.get(19).getBollUp());
+        assertDecimal("-1.032563", values.get(19).getBollDn());
+
+        assertDecimal("1.000000", values.getFirst().getEma6());
+        assertDecimal("1.285714", values.get(1).getEma6());
+        assertDecimal("1.153846", values.get(1).getEma12());
+        assertDecimal("1.095238", values.get(1).getEma20());
+        assertDecimal("0.000000", values.getFirst().getMacdDif());
+        assertDecimal("0.000000", values.getFirst().getMacdDea());
+        assertDecimal("0.000000", values.getFirst().getMacdBar());
+        assertDecimal("0.079772", values.get(1).getMacdDif());
+        assertDecimal("0.015954", values.get(1).getMacdDea());
+        assertDecimal("0.127635", values.get(1).getMacdBar());
+    }
+
+    @Test
+    void calculatesSimpleMovingAverageRsiAndHandlesFlatPrices() {
+        List<KLineIndicator> mixed = calculator.calculate(records(
+                1, 2, 1, 3, 2, 4, 3, 5, 4, 6, 5, 7, 6), 1);
+
+        assertNull(mixed.get(5).getRsi6());
+        assertDecimal("62.5000", mixed.get(6).getRsi6());
+        assertNull(mixed.get(11).getRsi12());
+        assertDecimal("64.7059", mixed.get(12).getRsi12());
+
+        List<KLineIndicator> flat = calculator.calculate(records(
+                10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10, 10), 1);
+        assertDecimal("50.0000", flat.get(6).getRsi6());
+        assertDecimal("50.0000", flat.get(12).getRsi12());
+    }
+
+    @Test
+    void initializesAndSmoothsKdjFromFifty() {
+        List<KLineIndicator> values = calculator.calculate(records(1, 2), 1);
+
+        assertDecimal("50.0000", values.getFirst().getKdjK());
+        assertDecimal("50.0000", values.getFirst().getKdjD());
+        assertDecimal("50.0000", values.getFirst().getKdjJ());
+        assertDecimal("55.5556", values.get(1).getKdjK());
+        assertDecimal("51.8519", values.get(1).getKdjD());
+        assertDecimal("62.9630", values.get(1).getKdjJ());
+    }
+
+    @Test
+    void incrementalCalculationMatchesFullHistoryAndOnlyReturnsNewRows() {
+        List<KLineRecord> all = sequence(1, 70);
+        List<KLineIndicator> full = calculator.calculate(all, 1);
+        List<KLineRecord> history = all.subList(0, 60);
+        List<KLineRecord> additions = all.subList(60, 70);
+
+        List<KLineIndicator> incremental = calculator.calculateIncremental(
+                history.subList(history.size() - 60, history.size()),
+                additions,
+                full.get(59),
+                1);
+
+        assertEquals(10, incremental.size());
+        for (int i = 0; i < incremental.size(); i++) {
+            KLineIndicator expected = full.get(60 + i);
+            KLineIndicator actual = incremental.get(i);
+            assertEquals(expected.getKTime(), actual.getKTime());
+            assertEquals(expected.getMa60(), actual.getMa60());
+            assertEquals(expected.getBollUp(), actual.getBollUp());
+            assertClose(expected.getEma20(), actual.getEma20(), new BigDecimal("0.000002"));
+            assertClose(expected.getMacdBar(), actual.getMacdBar(), new BigDecimal("0.000002"));
+            assertEquals(expected.getRsi12(), actual.getRsi12());
+            assertClose(expected.getKdjJ(), actual.getKdjJ(), new BigDecimal("0.0002"));
+        }
+    }
+
+    @Test
+    void incrementalMacdUsesPreviousDifWhenPreviousDeaIsMissing() {
+        KLineIndicator previous = KLineIndicator.builder()
+                .instrumentId(7)
+                .kTime(1L)
+                .ema6(new BigDecimal("10"))
+                .ema12(new BigDecimal("10"))
+                .ema20(new BigDecimal("10"))
+                .macdDif(new BigDecimal("2"))
+                .macdDea(null)
+                .kdjK(new BigDecimal("50"))
+                .kdjD(new BigDecimal("50"))
+                .build();
+
+        KLineIndicator actual = calculator.calculateIncremental(
+                List.of(),
+                records(13),
+                previous,
+                1).getFirst();
+
+        assertDecimal("2.018234", actual.getMacdDea());
+    }
+
+    private List<KLineRecord> sequence(int from, int to) {
+        List<Integer> closes = new ArrayList<>();
+        for (int value = from; value <= to; value++) {
+            closes.add(value);
+        }
+        return records(closes.stream().mapToInt(Integer::intValue).toArray());
+    }
+
+    private List<KLineRecord> records(int... closes) {
+        List<KLineRecord> records = new ArrayList<>();
+        for (int i = 0; i < closes.length; i++) {
+            int close = closes[i];
+            records.add(KLineRecord.builder()
+                    .instrumentId(7)
+                    .kTime((long) i + 1)
+                    .open(close)
+                    .high(close + 1)
+                    .low(close - 1)
+                    .close(close)
+                    .volume(100L)
+                    .turnover(1000L)
+                    .openInterest(50L)
+                    .build());
+        }
+        return records;
+    }
+
+    private void assertDecimal(String expected, BigDecimal actual) {
+        assertEquals(new BigDecimal(expected), actual);
+    }
+
+    private void assertClose(BigDecimal expected, BigDecimal actual, BigDecimal tolerance) {
+        assertTrue(expected.subtract(actual).abs().compareTo(tolerance) <= 0,
+                () -> "expected " + expected + " but was " + actual);
+    }
+}
diff --git a/src/test/java/com/yangwale/backtestify/service/market/repository/KLineTableResolverTest.java b/src/test/java/com/yangwale/backtestify/service/market/repository/KLineTableResolverTest.java
index 8435da6..2c62d8d 100644
--- a/src/test/java/com/yangwale/backtestify/service/market/repository/KLineTableResolverTest.java
+++ b/src/test/java/com/yangwale/backtestify/service/market/repository/KLineTableResolverTest.java
@@ -13,6 +13,7 @@ class KLineTableResolverTest {
     @Test
     void resolvesSupportedRealMarketPeriods() {
         assertEquals("t_kline_1m", resolver.resolve("1m"));
+        assertEquals("t_kline_3m", resolver.resolve("3m"));
         assertEquals("t_kline_5m", resolver.resolve("5m"));
         assertEquals("t_kline_15m", resolver.resolve("15m"));
         assertEquals("t_kline_30m", resolver.resolve("30m"));
@@ -20,10 +21,15 @@ class KLineTableResolverTest {
         assertEquals("t_kline_4h", resolver.resolve("4h"));
         assertEquals("t_kline_1d", resolver.resolve("1d"));
         assertEquals("t_kline_1w", resolver.resolve("1w"));
+        assertEquals("t_kline_1mo", resolver.resolve("1mo"));
+        assertEquals("t_indicator_1m", resolver.resolveIndicator("1m"));
+        assertEquals("t_indicator_3m", resolver.resolveIndicator("3m"));
+        assertEquals("t_indicator_1d", resolver.resolveIndicator("1d"));
+        assertEquals("t_indicator_1mo", resolver.resolveIndicator("1mo"));
     }
 
     @Test
-    void rejects3mBecauseRealMarketDoesNotProvideIt() {
-        assertThrows(BusinessException.class, () -> resolver.resolve("3m"));
+    void rejectsUnknownPeriod() {
+        assertThrows(BusinessException.class, () -> resolver.resolve("2h"));
     }
 }
diff --git a/src/test/java/com/yangwale/backtestify/service/market/sync/MarketDataSyncServiceTest.java b/src/test/java/com/yangwale/backtestify/service/market/sync/MarketDataSyncServiceTest.java
new file mode 100644
index 0000000..c79329c
--- /dev/null
+++ b/src/test/java/com/yangwale/backtestify/service/market/sync/MarketDataSyncServiceTest.java
@@ -0,0 +1,85 @@
+package com.yangwale.backtestify.service.market.sync;
+
+import com.baomidou.mybatisplus.core.conditions.Wrapper;
+import com.yangwale.backtestify.config.MarketDataProperties;
+import com.yangwale.backtestify.entity.InstrumentDictionary;
+import com.yangwale.backtestify.entity.KLineRecord;
+import com.yangwale.backtestify.entity.MarketDataSyncLog;
+import com.yangwale.backtestify.mapper.InstrumentDictionaryMapper;
+import com.yangwale.backtestify.mapper.KLineMapper;
+import com.yangwale.backtestify.mapper.MarketDataSyncLogMapper;
+import com.yangwale.backtestify.service.market.client.CnQuotationClient;
+import com.yangwale.backtestify.service.market.client.CnQuotationModels;
+import com.yangwale.backtestify.service.market.indicator.IndicatorCalculationService;
+import com.yangwale.backtestify.service.market.repository.KLineTableResolver;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class MarketDataSyncServiceTest {
+
+    @Test
+    void calculatesIncrementalIndicatorsAfterKLinesArePersisted() {
+        MarketDataProperties properties = new MarketDataProperties();
+        properties.getSync().setPeriods(List.of("1d"));
+        InstrumentDictionaryMapper instrumentMapper = mock(InstrumentDictionaryMapper.class);
+        KLineMapper kLineMapper = mock(KLineMapper.class);
+        MarketDataSyncLogMapper syncLogMapper = mock(MarketDataSyncLogMapper.class);
+        IndicatorCalculationService indicatorService = mock(IndicatorCalculationService.class);
+        InstrumentDictionary instrument = new InstrumentDictionary();
+        instrument.setId(7);
+        instrument.setExchangeId("SHFE");
+        instrument.setContractCode("rb2610");
+        instrument.setPriceScale(100);
+        instrument.setIsMain(1);
+        when(instrumentMapper.selectList(any(Wrapper.class))).thenReturn(List.of(instrument));
+        when(kLineMapper.upsertBatch(eq("t_kline_1d"), anyList())).thenReturn(1);
+
+        MarketDataSyncService service = new MarketDataSyncService(
+                properties,
+                new FakeQuotationClient(properties),
+                instrumentMapper,
+                kLineMapper,
+                syncLogMapper,
+                new KLineTableResolver(),
+                indicatorService) {
+            @Override
+            public void refreshMainContracts() {
+                // Contract refresh is outside this test's synchronization seam.
+            }
+        };
+
+        service.syncIncrementalForMainContracts();
+
+        verify(kLineMapper).upsertBatch(eq("t_kline_1d"), anyList());
+        verify(indicatorService).updateIncremental("1d", 7);
+        verify(syncLogMapper, org.mockito.Mockito.atLeastOnce()).insert(any(MarketDataSyncLog.class));
+    }
+
+    private static final class FakeQuotationClient extends CnQuotationClient {
+
+        private FakeQuotationClient(MarketDataProperties properties) {
+            super(properties);
+        }
+
+        @Override
+        public List<CnQuotationModels.GoodsItem> listMainContracts() {
+            return List.of();
+        }
+
+        @Override
+        public List<CnQuotationModels.KChartItem> getKChartByDate(
+                String excode, String code, String period, long date, String direction) {
+            return List.of(new CnQuotationModels.KChartItem(
+                    null, "100.00", "101.00", "99.00", "100.50",
+                    "10", "1000", date + 1, "20", null));
+        }
+    }
+}
diff --git a/src/test/resources/schema-test.sql b/src/test/resources/schema-test.sql
index c8e570b..3f7f6a2 100644
--- a/src/test/resources/schema-test.sql
+++ b/src/test/resources/schema-test.sql
@@ -14,7 +14,7 @@ CREATE TABLE IF NOT EXISTS t_instrument_dictionary (
 
 CREATE TABLE IF NOT EXISTS t_kline_1m (
     instrument_id SMALLINT NOT NULL,
-    timestamp     INT NOT NULL,
+    k_time        INT NOT NULL,
     open          INT NOT NULL,
     high          INT NOT NULL,
     low           INT NOT NULL,
@@ -22,11 +22,11 @@ CREATE TABLE IF NOT EXISTS t_kline_1m (
     volume        INT NOT NULL DEFAULT 0,
     turnover      BIGINT NOT NULL DEFAULT 0,
     open_interest INT NOT NULL DEFAULT 0,
-    PRIMARY KEY (instrument_id, timestamp)
+    PRIMARY KEY (instrument_id, k_time)
 );
 CREATE TABLE IF NOT EXISTS t_kline_5m (
     instrument_id SMALLINT NOT NULL,
-    timestamp     INT NOT NULL,
+    k_time        INT NOT NULL,
     open          INT NOT NULL,
     high          INT NOT NULL,
     low           INT NOT NULL,
@@ -34,11 +34,11 @@ CREATE TABLE IF NOT EXISTS t_kline_5m (
     volume        INT NOT NULL DEFAULT 0,
     turnover      BIGINT NOT NULL DEFAULT 0,
     open_interest INT NOT NULL DEFAULT 0,
-    PRIMARY KEY (instrument_id, timestamp)
+    PRIMARY KEY (instrument_id, k_time)
 );
 CREATE TABLE IF NOT EXISTS t_kline_15m (
     instrument_id SMALLINT NOT NULL,
-    timestamp     INT NOT NULL,
+    k_time        INT NOT NULL,
     open          INT NOT NULL,
     high          INT NOT NULL,
     low           INT NOT NULL,
@@ -46,11 +46,11 @@ CREATE TABLE IF NOT EXISTS t_kline_15m (
     volume        INT NOT NULL DEFAULT 0,
     turnover      BIGINT NOT NULL DEFAULT 0,
     open_interest INT NOT NULL DEFAULT 0,
-    PRIMARY KEY (instrument_id, timestamp)
+    PRIMARY KEY (instrument_id, k_time)
 );
 CREATE TABLE IF NOT EXISTS t_kline_30m (
     instrument_id SMALLINT NOT NULL,
-    timestamp     INT NOT NULL,
+    k_time        INT NOT NULL,
     open          INT NOT NULL,
     high          INT NOT NULL,
     low           INT NOT NULL,
@@ -58,11 +58,11 @@ CREATE TABLE IF NOT EXISTS t_kline_30m (
     volume        INT NOT NULL DEFAULT 0,
     turnover      BIGINT NOT NULL DEFAULT 0,
     open_interest INT NOT NULL DEFAULT 0,
-    PRIMARY KEY (instrument_id, timestamp)
+    PRIMARY KEY (instrument_id, k_time)
 );
 CREATE TABLE IF NOT EXISTS t_kline_1h (
     instrument_id SMALLINT NOT NULL,
-    timestamp     INT NOT NULL,
+    k_time        INT NOT NULL,
     open          INT NOT NULL,
     high          INT NOT NULL,
     low           INT NOT NULL,
@@ -70,11 +70,11 @@ CREATE TABLE IF NOT EXISTS t_kline_1h (
     volume        INT NOT NULL DEFAULT 0,
     turnover      BIGINT NOT NULL DEFAULT 0,
     open_interest INT NOT NULL DEFAULT 0,
-    PRIMARY KEY (instrument_id, timestamp)
+    PRIMARY KEY (instrument_id, k_time)
 );
 CREATE TABLE IF NOT EXISTS t_kline_4h (
     instrument_id SMALLINT NOT NULL,
-    timestamp     INT NOT NULL,
+    k_time        INT NOT NULL,
     open          INT NOT NULL,
     high          INT NOT NULL,
     low           INT NOT NULL,
@@ -82,11 +82,11 @@ CREATE TABLE IF NOT EXISTS t_kline_4h (
     volume        INT NOT NULL DEFAULT 0,
     turnover      BIGINT NOT NULL DEFAULT 0,
     open_interest INT NOT NULL DEFAULT 0,
-    PRIMARY KEY (instrument_id, timestamp)
+    PRIMARY KEY (instrument_id, k_time)
 );
 CREATE TABLE IF NOT EXISTS t_kline_1d (
     instrument_id SMALLINT NOT NULL,
-    timestamp     INT NOT NULL,
+    k_time        INT NOT NULL,
     open          INT NOT NULL,
     high          INT NOT NULL,
     low           INT NOT NULL,
@@ -94,11 +94,11 @@ CREATE TABLE IF NOT EXISTS t_kline_1d (
     volume        INT NOT NULL DEFAULT 0,
     turnover      BIGINT NOT NULL DEFAULT 0,
     open_interest INT NOT NULL DEFAULT 0,
-    PRIMARY KEY (instrument_id, timestamp)
+    PRIMARY KEY (instrument_id, k_time)
 );
 CREATE TABLE IF NOT EXISTS t_kline_1w (
     instrument_id SMALLINT NOT NULL,
-    timestamp     INT NOT NULL,
+    k_time        INT NOT NULL,
     open          INT NOT NULL,
     high          INT NOT NULL,
     low           INT NOT NULL,
@@ -106,9 +106,41 @@ CREATE TABLE IF NOT EXISTS t_kline_1w (
     volume        INT NOT NULL DEFAULT 0,
     turnover      BIGINT NOT NULL DEFAULT 0,
     open_interest INT NOT NULL DEFAULT 0,
-    PRIMARY KEY (instrument_id, timestamp)
+    PRIMARY KEY (instrument_id, k_time)
 );
 
+CREATE TABLE IF NOT EXISTS t_indicator_1m (
+    instrument_id SMALLINT NOT NULL,
+    k_time        INT NOT NULL,
+    ma5           DECIMAL(18,6),
+    ma10          DECIMAL(18,6),
+    ma20          DECIMAL(18,6),
+    ma60          DECIMAL(18,6),
+    boll_mb       DECIMAL(18,6),
+    boll_up       DECIMAL(18,6),
+    boll_dn       DECIMAL(18,6),
+    ema6          DECIMAL(18,6),
+    ema12         DECIMAL(18,6),
+    ema20         DECIMAL(18,6),
+    macd_dif      DECIMAL(18,6),
+    macd_dea      DECIMAL(18,6),
+    macd_bar      DECIMAL(18,6),
+    rsi6          DECIMAL(10,4),
+    rsi12         DECIMAL(10,4),
+    kdj_k         DECIMAL(10,4),
+    kdj_d         DECIMAL(10,4),
+    kdj_j         DECIMAL(10,4),
+    create_at     INT,
+    PRIMARY KEY (instrument_id, k_time)
+);
+CREATE TABLE IF NOT EXISTS t_indicator_5m AS SELECT * FROM t_indicator_1m WHERE 1 = 0;
+CREATE TABLE IF NOT EXISTS t_indicator_15m AS SELECT * FROM t_indicator_1m WHERE 1 = 0;
+CREATE TABLE IF NOT EXISTS t_indicator_30m AS SELECT * FROM t_indicator_1m WHERE 1 = 0;
+CREATE TABLE IF NOT EXISTS t_indicator_1h AS SELECT * FROM t_indicator_1m WHERE 1 = 0;
+CREATE TABLE IF NOT EXISTS t_indicator_4h AS SELECT * FROM t_indicator_1m WHERE 1 = 0;
+CREATE TABLE IF NOT EXISTS t_indicator_1d AS SELECT * FROM t_indicator_1m WHERE 1 = 0;
+CREATE TABLE IF NOT EXISTS t_indicator_1w AS SELECT * FROM t_indicator_1m WHERE 1 = 0;
+
 CREATE TABLE IF NOT EXISTS t_market_data_sync_log (
     id            BIGINT AUTO_INCREMENT PRIMARY KEY,
     sync_type     VARCHAR(32) NOT NULL,