Lee/backtestify
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add persisted technical indicator calculation

Browse Source
This commit is contained in:
Lee 2026-07-24 12:56:20 +08:00
parent 61dd0073d5
commit b5e34b6d18
17 changed files with 1095 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
src/main/java/com/yangwale/backtestify/entity/KLineIndicator.java Normal file
View File

@ -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;
}

2
src/main/java/com/yangwale/backtestify/entity/KLineRecord.java
View File

@ -15,7 +15,7 @@ public class KLineRecord {
private Integer instrumentId; private Integer instrumentId;
/** Unix时间戳，秒级 */ /** Unix时间戳，秒级 */
private Long timestamp; private Long kTime;
private Integer open; private Integer open;

77
src/main/java/com/yangwale/backtestify/mapper/IndicatorMapper.java Normal file
View File

@ -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);
}

46
src/main/java/com/yangwale/backtestify/mapper/KLineMapper.java
View File

@ -12,25 +12,61 @@ import java.util.List;
public interface KLineMapper { public interface KLineMapper {
@Select(""" @Select("""
SELECT instrument_id, timestamp, open, high, low, close, SELECT instrument_id, k_time, open, high, low, close,
volume, turnover, open_interest volume, turnover, open_interest
FROM ${tableName} FROM ${tableName}
WHERE instrument_id = #{instrumentId} WHERE instrument_id = #{instrumentId}
AND timestamp BETWEEN #{startTimestamp} AND #{endTimestamp} AND k_time BETWEEN #{startTimestamp} AND #{endTimestamp}
ORDER BY timestamp ASC ORDER BY k_time ASC
""") """)
List<KLineRecord> selectRange(@Param("tableName") String tableName, List<KLineRecord> selectRange(@Param("tableName") String tableName,
@Param("instrumentId") Integer instrumentId, @Param("instrumentId") Integer instrumentId,
@Param("startTimestamp") long startTimestamp, @Param("startTimestamp") long startTimestamp,
@Param("endTimestamp") long endTimestamp); @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(""" @Insert("""
<script> <script>
INSERT INTO ${tableName} 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 VALUES
<foreach collection="records" item="item" separator=","> <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}) #{item.volume}, #{item.turnover}, #{item.openInterest})
</foreach> </foreach>
ON DUPLICATE KEY UPDATE ON DUPLICATE KEY UPDATE

2
src/main/java/com/yangwale/backtestify/service/market/MysqlMarketDataProvider.java
View File

@ -84,7 +84,7 @@ public class MysqlMarketDataProvider implements MarketDataProvider {
private KLineData toKLineData(KLineRecord record, int priceScale) { private KLineData toKLineData(KLineRecord record, int priceScale) {
return KLineData.builder() 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)) .open(PriceScaleConverter.toRaw(record.getOpen(), priceScale))
.high(PriceScaleConverter.toRaw(record.getHigh(), priceScale)) .high(PriceScaleConverter.toRaw(record.getHigh(), priceScale))
.low(PriceScaleConverter.toRaw(record.getLow(), priceScale)) .low(PriceScaleConverter.toRaw(record.getLow(), priceScale))

21
src/main/java/com/yangwale/backtestify/service/market/indicator/IndicatorCalculationService.java Normal file
View File

@ -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);
}

19
src/main/java/com/yangwale/backtestify/service/market/indicator/IndicatorCalculator.java Normal file
View File

@ -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);
}

275
src/main/java/com/yangwale/backtestify/service/market/indicator/TechnicalIndicatorCalculator.java Normal file
View File

@ -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);
}
}
}

92
src/main/java/com/yangwale/backtestify/service/market/indicator/impl/IndicatorCalculationServiceImpl.java Normal file
View File

@ -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;
}
}

9
src/main/java/com/yangwale/backtestify/service/market/repository/KLineTableResolver.java
View File

@ -15,13 +15,15 @@ public class KLineTableResolver {
private static final Map<String, String> TABLES = Map.of( private static final Map<String, String> TABLES = Map.of(
"1m", "t_kline_1m", "1m", "t_kline_1m",
"3m", "t_kline_3m",
"5m", "t_kline_5m", "5m", "t_kline_5m",
"15m", "t_kline_15m", "15m", "t_kline_15m",
"30m", "t_kline_30m", "30m", "t_kline_30m",
"1h", "t_kline_1h", "1h", "t_kline_1h",
"4h", "t_kline_4h", "4h", "t_kline_4h",
"1d", "t_kline_1d", "1d", "t_kline_1d",
"1w", "t_kline_1w" "1w", "t_kline_1w",
"1mo", "t_kline_1mo"
); );
public String resolve(String period) { public String resolve(String period) {
@ -33,6 +35,11 @@ public class KLineTableResolver {
return tableName; return tableName;
} }
public String resolveIndicator(String period) {
String kLineTable = resolve(period);
return kLineTable.replace("t_kline_", "t_indicator_");
}
public Set<String> supportedPeriods() { public Set<String> supportedPeriods() {
return TABLES.keySet(); return TABLES.keySet();
} }

5
src/main/java/com/yangwale/backtestify/service/market/sync/MarketDataSyncService.java
View File

@ -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.CnQuotationClient;
import com.yangwale.backtestify.service.market.client.CnQuotationModels; import com.yangwale.backtestify.service.market.client.CnQuotationModels;
import com.yangwale.backtestify.service.market.convert.PriceScaleConverter; 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 com.yangwale.backtestify.service.market.repository.KLineTableResolver;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
@ -41,6 +42,7 @@ public class MarketDataSyncService {
private final KLineMapper kLineMapper; private final KLineMapper kLineMapper;
private final MarketDataSyncLogMapper syncLogMapper; private final MarketDataSyncLogMapper syncLogMapper;
private final KLineTableResolver tableResolver; private final KLineTableResolver tableResolver;
private final IndicatorCalculationService indicatorCalculationService;
@Scheduled(cron = "${market-data.sync.cron:0 0 6 * * ?}", zone = "${market-data.sync.zone:Asia/Shanghai}") @Scheduled(cron = "${market-data.sync.cron:0 0 6 * * ?}", zone = "${market-data.sync.zone:Asia/Shanghai}")
public void syncYesterdayMainContracts() { public void syncYesterdayMainContracts() {
@ -111,6 +113,7 @@ public class MarketDataSyncService {
.toList(); .toList();
if (!records.isEmpty()) { if (!records.isEmpty()) {
count = kLineMapper.upsertBatch(tableResolver.resolve(period), records); count = kLineMapper.upsertBatch(tableResolver.resolve(period), records);
indicatorCalculationService.updateIncremental(period, instrument.getId());
} }
saveLog("KLINE", period, instrument.getContractCode(), syncDate, "SUCCESS", count, null, start); saveLog("KLINE", period, instrument.getContractCode(), syncDate, "SUCCESS", count, null, start);
} catch (Exception e) { } catch (Exception e) {
@ -148,7 +151,7 @@ public class MarketDataSyncService {
try { try {
return KLineRecord.builder() return KLineRecord.builder()
.instrumentId(instrument.getId()) .instrumentId(instrument.getId())
.timestamp(item.u()) .kTime(item.u())
.open(PriceScaleConverter.toScaled(item.o(), instrument.getPriceScale())) .open(PriceScaleConverter.toScaled(item.o(), instrument.getPriceScale()))
.high(PriceScaleConverter.toScaled(item.h(), instrument.getPriceScale())) .high(PriceScaleConverter.toScaled(item.h(), instrument.getPriceScale()))
.low(PriceScaleConverter.toScaled(item.l(), instrument.getPriceScale())) .low(PriceScaleConverter.toScaled(item.l(), instrument.getPriceScale()))

125
src/test/java/com/yangwale/backtestify/service/market/indicator/IndicatorCalculationServiceImplTest.java Normal file
View File

@ -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();
}
}

82
src/test/java/com/yangwale/backtestify/service/market/indicator/IndicatorPersistenceIntegrationTest.java Normal file
View File

@ -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;
}
}

159
src/test/java/com/yangwale/backtestify/service/market/indicator/TechnicalIndicatorCalculatorTest.java Normal file
View File

@ -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);
}
}

10
src/test/java/com/yangwale/backtestify/service/market/repository/KLineTableResolverTest.java
View File

@ -13,6 +13,7 @@ class KLineTableResolverTest {
@Test @Test
void resolvesSupportedRealMarketPeriods() { void resolvesSupportedRealMarketPeriods() {
assertEquals("t_kline_1m", resolver.resolve("1m")); assertEquals("t_kline_1m", resolver.resolve("1m"));
assertEquals("t_kline_3m", resolver.resolve("3m"));
assertEquals("t_kline_5m", resolver.resolve("5m")); assertEquals("t_kline_5m", resolver.resolve("5m"));
assertEquals("t_kline_15m", resolver.resolve("15m")); assertEquals("t_kline_15m", resolver.resolve("15m"));
assertEquals("t_kline_30m", resolver.resolve("30m")); assertEquals("t_kline_30m", resolver.resolve("30m"));
@ -20,10 +21,15 @@ class KLineTableResolverTest {
assertEquals("t_kline_4h", resolver.resolve("4h")); assertEquals("t_kline_4h", resolver.resolve("4h"));
assertEquals("t_kline_1d", resolver.resolve("1d")); assertEquals("t_kline_1d", resolver.resolve("1d"));
assertEquals("t_kline_1w", resolver.resolve("1w")); 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 @Test
void rejects3mBecauseRealMarketDoesNotProvideIt() { void rejectsUnknownPeriod() {
assertThrows(BusinessException.class, () -> resolver.resolve("3m")); assertThrows(BusinessException.class, () -> resolver.resolve("2h"));
} }
} }

85
src/test/java/com/yangwale/backtestify/service/market/sync/MarketDataSyncServiceTest.java Normal file
View File

@ -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));
}
}
}

64
src/test/resources/schema-test.sql
View File

@ -14,7 +14,7 @@ CREATE TABLE IF NOT EXISTS t_instrument_dictionary (
CREATE TABLE IF NOT EXISTS t_kline_1m ( CREATE TABLE IF NOT EXISTS t_kline_1m (
instrument_id SMALLINT NOT NULL, instrument_id SMALLINT NOT NULL,
timestamp INT NOT NULL, k_time INT NOT NULL,
open INT NOT NULL, open INT NOT NULL,
high INT NOT NULL, high INT NOT NULL,
low 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, volume INT NOT NULL DEFAULT 0,
turnover BIGINT NOT NULL DEFAULT 0, turnover BIGINT NOT NULL DEFAULT 0,
open_interest INT 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 ( CREATE TABLE IF NOT EXISTS t_kline_5m (
instrument_id SMALLINT NOT NULL, instrument_id SMALLINT NOT NULL,
timestamp INT NOT NULL, k_time INT NOT NULL,
open INT NOT NULL, open INT NOT NULL,
high INT NOT NULL, high INT NOT NULL,
low 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, volume INT NOT NULL DEFAULT 0,
turnover BIGINT NOT NULL DEFAULT 0, turnover BIGINT NOT NULL DEFAULT 0,
open_interest INT 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 ( CREATE TABLE IF NOT EXISTS t_kline_15m (
instrument_id SMALLINT NOT NULL, instrument_id SMALLINT NOT NULL,
timestamp INT NOT NULL, k_time INT NOT NULL,
open INT NOT NULL, open INT NOT NULL,
high INT NOT NULL, high INT NOT NULL,
low 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, volume INT NOT NULL DEFAULT 0,
turnover BIGINT NOT NULL DEFAULT 0, turnover BIGINT NOT NULL DEFAULT 0,
open_interest INT 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 ( CREATE TABLE IF NOT EXISTS t_kline_30m (
instrument_id SMALLINT NOT NULL, instrument_id SMALLINT NOT NULL,
timestamp INT NOT NULL, k_time INT NOT NULL,
open INT NOT NULL, open INT NOT NULL,
high INT NOT NULL, high INT NOT NULL,
low 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, volume INT NOT NULL DEFAULT 0,
turnover BIGINT NOT NULL DEFAULT 0, turnover BIGINT NOT NULL DEFAULT 0,
open_interest INT 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 ( CREATE TABLE IF NOT EXISTS t_kline_1h (
instrument_id SMALLINT NOT NULL, instrument_id SMALLINT NOT NULL,
timestamp INT NOT NULL, k_time INT NOT NULL,
open INT NOT NULL, open INT NOT NULL,
high INT NOT NULL, high INT NOT NULL,
low 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, volume INT NOT NULL DEFAULT 0,
turnover BIGINT NOT NULL DEFAULT 0, turnover BIGINT NOT NULL DEFAULT 0,
open_interest INT 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 ( CREATE TABLE IF NOT EXISTS t_kline_4h (
instrument_id SMALLINT NOT NULL, instrument_id SMALLINT NOT NULL,
timestamp INT NOT NULL, k_time INT NOT NULL,
open INT NOT NULL, open INT NOT NULL,
high INT NOT NULL, high INT NOT NULL,
low 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, volume INT NOT NULL DEFAULT 0,
turnover BIGINT NOT NULL DEFAULT 0, turnover BIGINT NOT NULL DEFAULT 0,
open_interest INT 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 ( CREATE TABLE IF NOT EXISTS t_kline_1d (
instrument_id SMALLINT NOT NULL, instrument_id SMALLINT NOT NULL,
timestamp INT NOT NULL, k_time INT NOT NULL,
open INT NOT NULL, open INT NOT NULL,
high INT NOT NULL, high INT NOT NULL,
low 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, volume INT NOT NULL DEFAULT 0,
turnover BIGINT NOT NULL DEFAULT 0, turnover BIGINT NOT NULL DEFAULT 0,
open_interest INT 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 ( CREATE TABLE IF NOT EXISTS t_kline_1w (
instrument_id SMALLINT NOT NULL, instrument_id SMALLINT NOT NULL,
timestamp INT NOT NULL, k_time INT NOT NULL,
open INT NOT NULL, open INT NOT NULL,
high INT NOT NULL, high INT NOT NULL,
low 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, volume INT NOT NULL DEFAULT 0,
turnover BIGINT NOT NULL DEFAULT 0, turnover BIGINT NOT NULL DEFAULT 0,
open_interest INT 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 ( CREATE TABLE IF NOT EXISTS t_market_data_sync_log (
id BIGINT AUTO_INCREMENT PRIMARY KEY, id BIGINT AUTO_INCREMENT PRIMARY KEY,
sync_type VARCHAR(32) NOT NULL, sync_type VARCHAR(32) NOT NULL,