数据同步的沉默杀手：重复记录与回溯修改

凌晨 3:17，告警短信响起。

你揉着眼睛打开监控面板，发现定时任务报错——数据库连接超时。更糟糕的是，之前几次重试后的数据堆积在内存里，重复写入了几十万条记录。你花了两个小时手动清理，然后开始怀疑人生：为什么这么简单的事情，做了三次还是会出问题？

这不是你的代码写得烂。这是数据同步领域的核心矛盾：增量不增量，取决于你如何定义"增量"。

本文拆解三个实际问题：避免重复插入、追踪数据源修改、检测回溯更新。每一个都有可直接上线的生产级代码。

---

一、为什么"INSERT + DISTINCT"不是答案

很多团队的第一反应是：每次拉取后用 SQL 去重。

INSERT INTO kline_data (symbol, timestamp, open, high, low, close, volume)
SELECT symbol, timestamp, open, high, low, close, volume
FROM staging_data
WHERE NOT EXISTS (
    SELECT 1 FROM kline_data 
    WHERE kline_data.symbol = staging_data.symbol 
    AND kline_data.timestamp = staging_data.timestamp
);

这段代码能工作，但有三个致命问题：

问题	影响
全表扫描	1000 万条数据时，NOT EXISTS 子查询可能跑几分钟
事后补救	重复数据已经写入，再删再做性能浪费
无变更检测	数据源修改了某条记录，你无法感知

真正的增量同步应该在写入前就判断"这条数据是否需要处理"，而不是写入后再亡羊补牢。

---

二、UPSERT：让写入本身具备幂等性

2.1 核心思路

UPSERT (Update or Insert) 的本质是：用唯一约束定义"同一性"，数据库自动处理冲突。

-- PostgreSQL 语法
INSERT INTO kline_data (symbol, timestamp, open, high, low, close, volume, updated_at)
VALUES ('AAPL.US', '2026-01-15 09:30:00', 185.50, 186.20, 185.10, 185.90, 1250000, NOW())
ON CONFLICT (symbol, timestamp) 
DO UPDATE SET 
    open = EXCLUDED.open,
    high = EXCLUDED.high,
    low = EXCLUDED.low,
    close = EXCLUDED.close,
    volume = EXCLUDED.volume,
    updated_at = NOW();

关键点：

• (symbol, timestamp) 是复合唯一约束，定义了一条 K 线的"身份证"
• ON CONFLICT 后的 DO UPDATE 会在冲突时更新字段
• updated_at 记录本地数据库的更新时间

2.2 约束的创建与维护

-- 确保约束存在（幂等操作）
ALTER TABLE kline_data 
ADD CONSTRAINT kline_data_pkey UNIQUE (symbol, timestamp);

-- 索引优化：加速按 symbol + 时间范围的查询
CREATE INDEX idx_kline_symbol_time 
ON kline_data (symbol, timestamp DESC);

2.3 Python 生产级 UPSERT 实现

import os
import time
import psycopg2
from psycopg2.extras import execute_values
from typing import List, Dict, Any
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


class KlineUpsertManager:
    """K线数据 UPSERT 管理器 - 生产级"""
    
    def __init__(self):
        self.conn = psycopg2.connect(
            host=os.environ.get("DB_HOST", "localhost"),
            port=int(os.environ.get("DB_PORT", 5432)),
            database=os.environ.get("DB_NAME"),
            user=os.environ.get("DB_USER"),
            password=os.environ.get("DB_PASSWORD"),
            connect_timeout=10
        )
        self.conn.autocommit = False
    
    def ensure_constraints(self):
        """幂等创建约束和索引"""
        with self.conn.cursor() as cur:
            # 复合唯一约束
            cur.execute("""
                DO $$
                BEGIN
                    IF NOT EXISTS (
                        SELECT 1 FROM pg_constraint 
                        WHERE conname = 'kline_data_symbol_timestamp_key'
                    ) THEN
                        ALTER TABLE kline_data 
                        ADD CONSTRAINT kline_data_symbol_timestamp_key 
                        UNIQUE (symbol, timestamp);
                    END IF;
                END $$;
            """)
            
            # 索引（如果不存在）
            cur.execute("""
                CREATE INDEX IF NOT EXISTS idx_kline_symbol_time 
                ON kline_data (symbol, timestamp DESC);
            """)
        self.conn.commit()
        logger.info("约束和索引检查完成")
    
    def upsert_batch(self, records: List[Dict[str, Any]], batch_size: int = 1000):
        """
        批量 UPSERT K线数据
        
        Args:
            records: [{"symbol": "AAPL.US", "timestamp": "2026-01-15 09:30:00", 
                       "open": 185.50, "high": 186.20, "low": 185.10, 
                       "close": 185.90, "volume": 1250000}, ...]
            batch_size: 每批次处理条数
        """
        if not records:
            return 0
        
        columns = ["symbol", "timestamp", "open", "high", "low", "close", "volume"]
        values = [[r[col] for col in columns] for r in records]
        
        total_upserted = 0
        for i in range(0, len(values), batch_size):
            batch = values[i:i + batch_size]
            
            try:
                with self.conn.cursor() as cur:
                    execute_values(
                        cur,
                        """
                        INSERT INTO kline_data (symbol, timestamp, open, high, low, close, volume, updated_at)
                        VALUES %s
                        ON CONFLICT (symbol, timestamp) 
                        DO UPDATE SET 
                            open = EXCLUDED.open,
                            high = EXCLUDED.high,
                            low = EXCLUDED.low,
                            close = EXCLUDED.close,
                            volume = EXCLUDED.volume,
                            updated_at = NOW()
                        """,
                        batch,
                        page_size=batch_size
                    )
                self.conn.commit()
                total_upserted += len(batch)
                logger.info(f"批次 {i // batch_size + 1}: 成功 UPSERT {len(batch)} 条")
                
            except psycopg2.Error as e:
                self.conn.rollback()
                logger.error(f"批次写入失败，回滚: {e}")
                # ⚠️ 生产环境应写入死信队列或重试有限次
                raise
        
        return total_upserted
    
    def close(self):
        self.conn.close()

使用示例：

# 初始化并确保约束存在
manager = KlineUpsertManager()
manager.ensure_constraints()

# 模拟从 TickDB 获取的 K线数据
kline_records = [
    {"symbol": "AAPL.US", "timestamp": "2026-01-15 09:30:00", 
     "open": 185.50, "high": 186.20, "low": 185.10, "close": 185.90, "volume": 1250000},
    {"symbol": "AAPL.US", "timestamp": "2026-01-15 09:31:00", 
     "open": 185.90, "high": 186.50, "low": 185.80, "close": 186.20, "volume": 980000},
    # 即使 AAPL.US 09:30:00 重复推送，也不会创建两条记录
]

upserted = manager.upsert_batch(kline_records)
print(f"本次共写入/更新 {upserted} 条记录")

manager.close()

UPSERT 的核心优势：

• 写入本身幂等，多次执行结果一致
• 数据库层面原子处理，无竞态条件
• 批量处理性能高，10万条/秒是常见基准

---

三、版本号机制：让回溯修改无处遁形

3.1 什么是"回溯修改"

数据源回溯修改是指：某个时间点的数据，在之后被重新发布/修正。

典型场景：

• 上市公司财报后，临时发布修正公告
• 交易所对历史成交记录进行清算调整
• K线数据供应商对早期数据进行复权处理

如果只用 (symbol, timestamp) 作为唯一标识，上述修改会被静默覆盖，你永远不知道数据曾经被改过。

3.2 版本号设计

解决方案：在唯一标识中引入版本维度。

-- 方案 A：版本号字段（推荐）
ALTER TABLE kline_data 
ADD COLUMN version INT DEFAULT 1;

ALTER TABLE kline_data 
ADD CONSTRAINT kline_data_pkey_v2 UNIQUE (symbol, timestamp, version);

-- 方案 B：数据指纹（适合需要审计的场景）
ALTER TABLE kline_data 
ADD COLUMN data_hash VARCHAR(64);  -- SHA-256(data_content)

方案 A 适用场景：版本变更有明确来源（如交易所公告版本号）

方案 B 适用场景：任何字段变化都需要被记录

3.3 增强版 UPSERT：检测版本变更

import hashlib
import json

def compute_data_hash(record: Dict[str, Any]) -> str:
    """计算数据内容指纹"""
    # 排除元数据字段
    content_fields = ["open", "high", "low", "close", "volume"]
    content = {k: record[k] for k in content_fields}
    return hashlib.sha256(json.dumps(content, sort_keys=True).encode()).hexdigest()


def upsert_with_version_detection(manager, records: List[Dict[str, Any]]):
    """
    UPSERT 并检测数据内容变更
    
    Returns:
        (upserted_count, updated_count, content_changed_count)
    """
    upserted = 0
    updated = 0
    content_changed = 0
    
    for record in records:
        data_hash = compute_data_hash(record)
        
        # 检查是否已存在且内容一致
        with manager.conn.cursor() as cur:
            cur.execute("""
                SELECT id, data_hash FROM kline_data 
                WHERE symbol = %s AND timestamp = %s
                ORDER BY version DESC
                LIMIT 1
            """, (record["symbol"], record["timestamp"]))
            
            existing = cur.fetchone()
        
        if existing:
            old_id, old_hash = existing
            if old_hash != data_hash:
                # 数据内容变更，插入新版本
                record["version"] = 1  # 实际应为 SELECT MAX(version) + 1
                record["data_hash"] = data_hash
                updated += 1
                content_changed += 1
                logger.warning(
                    f"检测到数据变更: {record['symbol']}@{record['timestamp']} "
                    f"旧哈希 {old_hash[:8]} -> 新哈希 {data_hash[:8]}"
                )
        else:
            record["version"] = 1
            record["data_hash"] = data_hash
            upserted += 1
    
    # 批量写入（简化版）
    return upserted, updated, content_changed

3.4 版本回溯的查询策略

-- 获取某时间点最新有效数据
SELECT * FROM kline_data 
WHERE symbol = 'AAPL.US' 
  AND timestamp >= '2026-01-15 09:30:00' 
  AND timestamp < '2026-01-15 09:35:00'
ORDER BY timestamp, version DESC
LIMIT 1;

-- 追踪某条数据的所有历史版本（审计用）
SELECT * FROM kline_data 
WHERE symbol = 'AAPL.US' AND timestamp = '2026-01-15 09:30:00'
ORDER BY version DESC;

-- 统计发生了多少次回溯修改
SELECT 
    symbol,
    DATE(timestamp) as trade_date,
    COUNT(*) as total_records,
    SUM(CASE WHEN version > 1 THEN 1 ELSE 0 END) as revised_records
FROM kline_data
GROUP BY symbol, DATE(timestamp)
HAVING SUM(CASE WHEN version > 1 THEN 1 ELSE 0 END) > 0
ORDER BY revised_records DESC;

---

四、TickDB 增量拉取实战

4.1 与 TickDB 的数据交互模式

结合 TickDB 的 /kline 接口与本地 UPSERT 逻辑，构建完整的增量同步管道：

import os
import time
import requests
import psycopg2
from datetime import datetime, timedelta
from typing import Optional, List, Dict, Any


class TickDBKlineSyncer:
    """
    TickDB K线数据增量同步器
    支持断点续传、版本检测、限频处理
    """
    
    BASE_URL = "https://api.tickdb.ai/v1/market/kline"
    
    def __init__(self, db_manager: KlineUpsertManager):
        self.db = db_manager
        self.api_key = os.environ.get("TICKDB_API_KEY")
        self.headers = {"X-API-Key": self.api_key}
        self._last_sync_key: Optional[str] = None  # 断点续传标记
    
    def _fetch_klines(self, symbol: str, start_time: str, end_time: str, 
                      interval: str = "1m", limit: int = 1000) -> Dict[str, Any]:
        """
        从 TickDB 获取 K线数据
        
        ⚠️ 生产环境高频场景建议使用 aiohttp/asyncio
        """
        params = {
            "symbol": symbol,
            "interval": interval,
            "start_time": start_time,
            "end_time": end_time,
            "limit": limit
        }
        
        response = requests.get(
            self.BASE_URL,
            headers=self.headers,
            params=params,
            timeout=(3.05, 10)  # (connect_timeout, read_timeout)
        )
        
        # 限频处理
        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 60))
            print(f"触发限频，等待 {retry_after} 秒")
            time.sleep(retry_after)
            return self._fetch_klines(symbol, start_time, end_time, interval, limit)
        
        if response.status_code != 200:
            raise RuntimeError(f"API 请求失败: {response.status_code} - {response.text}")
        
        return response.json()
    
    def sync_symbol(self, symbol: str, days_back: int = 7, 
                    interval: str = "1m") -> Dict[str, int]:
        """
        同步单个交易品种
        
        Args:
            symbol: 交易品种代码，如 "AAPL.US"
            days_back: 向前回溯天数
            interval: K线周期
        
        Returns:
            {"upserted": N, "updated": M, "errors": E}
        """
        end_time = datetime.now()
        start_time = end_time - timedelta(days=days_back)
        
        start_str = start_time.strftime("%Y-%m-%d %H:%M:%S")
        end_str = end_time.strftime("%Y-%m-%d %H:%M:%S")
        
        print(f"开始同步 {symbol}: {start_str} -> {end_str}")
        
        total_stats = {"upserted": 0, "updated": 0, "errors": 0}
        page_count = 0
        
        while True:
            page_count += 1
            try:
                result = self._fetch_klines(symbol, start_str, end_str, interval)
                
                data = result.get("data", {}).get("klines", [])
                if not data:
                    print(f"  页面 {page_count}: 无数据，结束")
                    break
                
                # 转换为统一格式
                records = []
                for k in data:
                    records.append({
                        "symbol": symbol,
                        "timestamp": k["timestamp"],
                        "open": float(k["open"]),
                        "high": float(k["high"]),
                        "low": float(k["low"]),
                        "close": float(k["close"]),
                        "volume": float(k["volume"])
                    })
                
                # UPSERT 到本地数据库
                stats = self.db.upsert_batch(records)
                total_stats["upserted"] += stats
                
                # 断点续传：记录最新时间，下次从这里继续
                latest_timestamp = records[-1]["timestamp"]
                self._last_sync_key = latest_timestamp
                
                print(f"  页面 {page_count}: 写入 {stats} 条，最新时间 {latest_timestamp}")
                
                # 分页逻辑：如果返回 limit 条，说明还有下一页
                if len(data) < 1000:
                    break
                
                # 更新分页起点时间
                start_str = latest_timestamp
                time.sleep(0.1)  # ⚠️ 防止触发限频
                
            except Exception as e:
                print(f"  页面 {page_count} 出错: {e}")
                total_stats["errors"] += 1
                time.sleep(5)  # 指数退避会更好
        
        print(f"同步完成: {symbol} - {total_stats}")
        return total_stats
    
    def get_checkpoint(self, symbol: str) -> Optional[str]:
        """
        获取断点续传标记
        
        实际实现：从本地 Redis/文件读取上次同步的时间戳
        """
        # 简化实现：实际应从持久化存储读取
        return self._last_sync_key


def main():
    """每日定时同步任务入口"""
    
    # 初始化数据库管理器
    db_manager = KlineUpsertManager()
    db_manager.ensure_constraints()
    
    # 初始化 TickDB 同步器
    syncer = TickDBKlineSyncer(db_manager)
    
    # 需要同步的交易品种
    symbols = ["AAPL.US", "TSLA.US", "NVDA.US", "AMD.US"]
    
    total_stats = {"upserted": 0, "updated": 0, "errors": 0}
    
    for symbol in symbols:
        try:
            stats = syncer.sync_symbol(symbol, days_back=1)
            for k, v in stats.items():
                total_stats[k] += v
        except Exception as e:
            print(f"同步 {symbol} 失败: {e}")
    
    print(f"\n=== 同步任务完成 ===")
    print(f"总计: 写入 {total_stats['upserted']} 条, 更新 {total_stats['updated']} 条, 错误 {total_stats['errors']} 次")
    
    db_manager.close()


if __name__ == "__main__":
    main()

4.2 定时任务配置

使用 systemd 定时任务（推荐生产环境）：

# /etc/systemd/system/tickdb-sync.timer
[Unit]
Description=TickDB K线数据每日同步

[Timer]
OnCalendar=05:00:00
Persistent=true
RandomizedDelaySec=3600  # 随机延迟 0-1 小时，避免惊群

[Install]
WantedBy=timers.target

# /etc/systemd/system/tickdb-sync.service
[Unit]
Description=TickDB K线数据同步服务
After=network-online.target
Wants=network-online.target

[Service]
Type=oneshot
User=quant
Environment="PYTHONPATH=/opt/sync"
ExecStart=/usr/bin/python3 /opt/sync/sync_kline.py
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target

启动并验证：

sudo systemctl daemon-reload
sudo systemctl enable tickdb-sync.timer
sudo systemctl start tickdb-sync.timer
sudo systemctl status tickdb-sync.timer

---

五、完整方案对比

维度	纯 INSERT	INSERT + DISTINCT	UPSERT	UPSERT + 版本号
重复数据	❌ 产生	✅ 事后去重	✅ 写入即去重	✅ 写入即去重
回溯检测	❌ 无法感知	❌ 无法感知	❌ 静默覆盖	✅ 版本递增
写入性能	快	慢（全表扫描）	快（唯一索引）	快（唯一索引）
审计能力	无	无	弱	强
实现复杂度	低	中	中	高

推荐选择：

• 数据不常变更、只关心最新值 → UPSERT
• 需要变更追踪、审计回溯 → UPSERT + 版本号
• 数据量大、对性能敏感 → UPSERT + 批量写入

---

六、工程实践 Checklist

在生产环境中部署增量同步前，逐项检查：

检查项	说明	状态
唯一约束存在	ALTER TABLE ... ADD CONSTRAINT ... UNIQUE (symbol, timestamp)	⬜
索引已创建	(symbol, timestamp DESC) 复合索引	⬜
批量写入	使用 execute_values 而非逐条 INSERT	⬜
断点续传	同步失败后可从断点恢复	⬜
限频处理	API 返回 429 时读取 Retry-After	⬜
超时设置	HTTP 请求设置 timeout	⬜
事务边界	批量提交，避免长事务	⬜
监控告警	同步失败、延迟、数据量异常告警	⬜
版本检测（可选）	data_hash 追踪内容变更	⬜

---

结语

数据同步的核心难题不是"怎么拉数据"，而是如何定义"同一份数据"。

当 (symbol, timestamp) 是唯一键时，你关心的是"同一时刻的同一标的"；
当加入 version 或 data_hash 时，你关心的是"数据内容本身是否变化"。

TickDB 的 /kline 接口提供了时间范围查询能力，配合本地 UPSERT + 版本号机制，可以构建一套幂等、可审计、支持断点续传的增量同步管道。无论是日级别的数据归档，还是分钟级的实时监控，这套架构都能支撑。

---

下一步行动

如果你是个人开发者，想快速验证本文方案：

1. 访问 tickdb.ai 注册（免费，无需信用卡）
2. 在控制台生成 API Key
3. 安装 psycopg2（pip install psycopg2-binary），配置 PostgreSQL
4. 复制本文代码，设置环境变量后直接运行

如果你需要 10 年全量历史 K 线数据做离线回测，联系 [email protected] 了解 TickDB 历史数据导出服务，支持批量下载 + 自动分片。

如果你习惯用 AI 辅助开发，在 AI 助手中搜索安装 tickdb-market-data SKILL，用自然语言查询 K线数据并直接生成同步脚本。

---

本文不构成任何投资建议。市场有风险，投资需谨慎。