4. 标签定义与分类：监督学习标签、无监督学习标签、时序预测标签、事件驱动标签的定义与适用场景

// 以未来5分钟收益率作为回归标签
KStream<String, MarketTick> ticks = builder.stream("market.ticks");

KStream<String, LabeledTick> labeled = ticks
    .groupByKey()
    .windowedBy(TimeWindows.of(Duration.ofMinutes(5)).grace(Duration.ZERO))
    .aggregate(
        () -> new TickWindow(),
        (key, tick, window) -> window.add(tick),
        Materialized.<String, TickWindow, WindowStore<ByteArray, byte[]>>as("tick-window-store")
    )
    .toStream()
    .map((windowedKey, window) -> {
        double futureReturn = window.getFutureReturn(); // 窗口内最后价格 / 当前价格 - 1
        return new KeyValue<>(windowedKey.key(), new LabeledTick(window.getCurrentTick(), futureReturn));
    });

// 假设当前时间 t，构建输入窗口 [t-20, t] 和标签窗口 [t+1, t+10]
// 使用 Kafka Streams 的 SessionWindow 或自定义 Transformer

public class SequenceLabelTransformer implements Transformer<String, Tick, KeyValue<String, SequencePair>> {
    private StateStore store;

    @Override
    public KeyValue<String, SequencePair> transform(String key, Tick value) {
        // 将当前 tick 追加到环形缓冲区
        store.put(key, value);
        List<Tick> history = store.get(key); // 最近 30 个 tick
        if (history.size() == 30) {
            List<Double> inputSeq = history.subList(0, 20).stream().map(Tick::getPrice).collect(toList());
            List<Double> labelSeq = history.subList(20, 30).stream().map(Tick::getPrice).collect(toList());
            return new KeyValue<>(key, new SequencePair(inputSeq, labelSeq));
        }
        return null;
    }
}