feat: add offline data write-back and sync commands, bump to v1.6.0

- Add write_daily.py: encode/append daily bars to .day files
- Add write_ex_daily.py: encode/append extended market daily bars
- Add write_min_bar.py: encode/append minute bars (.5/.lc1/.lc5)
- Add sync-daily CLI: sync single stock with pagination support
- Add sync-all CLI: one-command sync for all SH/SZ .day files
- Update README with sync commands and Python write API docs
- 50 new unit tests covering encode round-trip, append dedup, edge cases
- Bump version 1.5.0 -> 1.6.0

src/easy_tdx/cli/cmd_offline.py

@@ -1,15 +1,25 @@
-"""离线本地数据读取命令（无需网络，读取本地通达信数据文件）。"""
+"""离线本地数据读写命令 —— 读取本地通达信数据文件 & 从服务端同步写入。"""
 
 from __future__ import annotations
 
+from pathlib import Path
+from typing import TYPE_CHECKING
+
 import click
 
+if TYPE_CHECKING:
+    import pandas as pd
+
+    from ..client import TdxClient
+    from ..models.bar import SecurityBar
+
 
 @click.group()
 def offline() -> None:
-    """离线本地数据读取（无需网络，读取本地通达信数据文件）。
+    """离线本地数据读写（读取本地通达信数据文件 & 从服务端同步写入）。
 
-    需要本地已安装通达信并下载过对应数据。
+    读取需要本地已安装通达信并下载过对应数据。
+    sync-daily 可从服务端获取最新日线并追加到本地 .day 文件。
 
     示例：
 
@@ -22,6 +32,10 @@ def offline() -> None:
       easy-tdx offline ex-files --table
 
       easy-tdx offline ex-daily 29#A1801 --table
+
+      easy-tdx offline sync-daily SZ 000001
+
+      easy-tdx offline sync-daily SH 600519 --vipdoc C:\\new_jyplug\\vipdoc
     """
     pass
 
@@ -412,3 +426,238 @@ def blocks(
         for b in result
     ]
     print_output(pd.DataFrame(rows), fmt)
+
+
+# ---------------------------------------------------------------------------
+# sync-daily：从服务端同步日线到本地 .day 文件
+# ---------------------------------------------------------------------------
+
+
+def _df_to_bars(df: pd.DataFrame) -> list[SecurityBar]:
+    """将日线 DataFrame 转换为 SecurityBar 列表（按日期升序）。"""
+    from ..models.bar import SecurityBar
+
+    bars: list[SecurityBar] = []
+    for _, row in df.iterrows():
+        dt = row["date"]
+        bars.append(
+            SecurityBar(
+                open=row["open"],
+                close=row["close"],
+                high=row["high"],
+                low=row["low"],
+                vol=row["vol"],
+                amount=row["amount"],
+                year=dt.year,
+                month=dt.month,
+                day=dt.day,
+                hour=0,
+                minute=0,
+            )
+        )
+    bars.sort(key=lambda b: b.year * 10000 + b.month * 100 + b.day)
+    return bars
+
+
+def _fetch_all_daily_bars(
+    client: TdxClient, market: int, code: str, need_full: bool = False
+) -> list[SecurityBar]:
+    """从服务端分页获取全部日线数据。
+
+    Args:
+        client: 已连接的 TdxClient。
+        market: 市场代码（0=SZ, 1=SH）。
+        code: 6 位股票代码。
+        need_full: True 表示拉取全量历史（空文件场景），
+                   False 表示只拉最近一页（增量更新）。
+
+    Returns:
+        SecurityBar 列表（按日期升序）。
+    """
+    from ..models.enums import KlineCategory
+
+    all_bars: list[SecurityBar] = []
+    start = 0
+    page_size = 800
+    max_pages = 50 if need_full else 1  # 50 页 = 40000 条，足够覆盖 A 股全部历史
+
+    for _ in range(max_pages):
+        df = client.get_security_bars(market, code, KlineCategory.DAY, start, page_size)
+        if df.empty:
+            break
+        all_bars.extend(_df_to_bars(df))
+        if len(df) < page_size:
+            break  # 最后一页不满，已无更多数据
+        start += page_size
+
+    # 去重并按日期升序排列（跨页可能有重叠）
+    seen: set[tuple[int, int, int]] = set()
+    unique: list[SecurityBar] = []
+    for b in all_bars:
+        key = (b.year, b.month, b.day)
+        if key not in seen:
+            seen.add(key)
+            unique.append(b)
+    unique.sort(key=lambda b: b.year * 10000 + b.month * 100 + b.day)
+    return unique
+
+
+def _sync_one_daily(client: TdxClient, filepath: Path) -> tuple[int, str]:
+    """同步单只股票日线，返回 (写入条数, 状态消息)。"""
+    from ..offline import append_daily_bars, get_last_bar_date
+    from ..offline.daily_bar import _SECURITY_COEFFICIENTS, _detect_security_type
+
+    # 文件名 → 市场 + 代码
+    name = filepath.name.lower()  # e.g. sh600000.day
+    exchange = name[:2]  # "sh" or "sz"
+    code = name[2:8]
+    market = 1 if exchange == "sh" else 0  # Market.SH=1, Market.SZ=0
+
+    # 判断是否需要全量拉取（空文件 → 全量，有数据 → 增量）
+    last_date = get_last_bar_date(filepath)
+    need_full = last_date is None
+
+    # 从服务端分页获取日线
+    bars = _fetch_all_daily_bars(client, market, code, need_full=need_full)
+    if not bars:
+        return 0, "服务端无数据"
+
+    # 检测证券类型获取系数
+    sec_type = _detect_security_type(filepath.name)
+    price_coeff, vol_coeff = _SECURITY_COEFFICIENTS.get(sec_type, (0.01, 0.01))
+
+    # 追加写入
+    written = append_daily_bars(filepath, bars, price_coeff, vol_coeff)
+    if written > 0:
+        return written, f"+{written}"
+    return 0, "已是最新"
+
+
+@offline.command("sync-daily")
+@click.argument("market")
+@click.argument("code")
+@click.option("--vipdoc", default=None, help="vipdoc 目录路径（默认自动检测）")
+def sync_daily(market: str, code: str, vipdoc: str | None) -> None:
+    """从服务端同步日线数据到本地 .day 文件。
+
+    自动检测本地文件末尾日期，从服务端分页获取缺失的数据并追加写入。
+    空文件自动全量下载，已有数据只做增量更新。
+    建议在通达信关闭时执行，避免文件被锁定。
+
+    MARKET: 市场代码（SZ/SH）
+    CODE: 6 位股票代码
+
+    示例：
+
+      easy-tdx offline sync-daily SZ 000001
+
+      easy-tdx offline sync-daily SH 000001 --vipdoc C:\\new_jyplug\\vipdoc
+    """
+    from ..client import TdxClient
+    from ..offline import find_daily_bar_file
+    from .output import print_error
+    from .parsers import parse_market
+
+    mkt = parse_market(market)
+
+    try:
+        filepath = find_daily_bar_file(mkt, code, vipdoc)
+        click.echo(f"目标文件: {filepath}")
+
+        click.echo("正在连接服务端获取日线数据...")
+        with TdxClient.from_best_host() as client:
+            written, msg = _sync_one_daily(client, filepath)
+
+        if written > 0:
+            click.echo(f"✓ 成功写入 {written} 条新记录")
+        else:
+            click.echo(f"本地已是最新，无需写入 ({msg})")
+
+    except PermissionError:
+        click.echo(f"✗ 文件被锁定，请关闭通达信后重试: {filepath}", err=True)
+        raise SystemExit(1)
+    except Exception as e:
+        print_error(str(e))
+        raise SystemExit(1)
+
+
+# ---------------------------------------------------------------------------
+# sync-all：一键同步全部日线
+# ---------------------------------------------------------------------------
+
+
+@offline.command("sync-all")
+@click.option("--vipdoc", default=None, help="vipdoc 目录路径（默认自动检测）")
+def sync_all(vipdoc: str | None) -> None:
+    """一键同步全部本地日线数据（沪深全市场）。
+
+    扫描 vipdoc 下所有 .day 文件，自动连接服务端获取最新数据并追加写入。
+    建议在通达信关闭时执行，避免文件被锁定。
+
+    示例：
+
+      easy-tdx offline sync-all
+
+      easy-tdx offline sync-all --vipdoc C:\\new_jyplug\\vipdoc
+    """
+    import time
+
+    from ..client import TdxClient
+    from ..offline.paths import resolve_vipdoc
+
+    try:
+        vipdoc_path = resolve_vipdoc(vipdoc)
+    except Exception as e:
+        click.echo(f"✗ {e}", err=True)
+        raise SystemExit(1)
+
+    # 1. 扫描所有 .day 文件
+    all_files: list[Path] = []
+    for exchange in ("sh", "sz"):
+        lday_dir = vipdoc_path / exchange / "lday"
+        if lday_dir.is_dir():
+            all_files.extend(sorted(lday_dir.glob("*.day")))
+
+    if not all_files:
+        click.echo("未找到任何 .day 文件，请确认 vipdoc 路径正确")
+        raise SystemExit(0)
+
+    total = len(all_files)
+    click.echo(f"发现 {total} 个 .day 文件，开始同步...")
+
+    # 2. 连接服务端，逐个同步
+    success = 0
+    skipped = 0
+    failed = 0
+    total_written = 0
+
+    with TdxClient.from_best_host() as client:
+        for idx, filepath in enumerate(all_files, 1):
+            name = filepath.name
+            try:
+                written, msg = _sync_one_daily(client, filepath)
+                total_written += written
+                if written > 0:
+                    success += 1
+                else:
+                    skipped += 1
+                click.echo(f"  [{idx}/{total}] {name}: {msg}")
+            except PermissionError:
+                failed += 1
+                click.echo(f"  [{idx}/{total}] {name}: ✗ 文件被锁定", err=True)
+            except Exception as e:
+                failed += 1
+                click.echo(f"  [{idx}/{total}] {name}: ✗ {e}", err=True)
+
+            # 每 100 只暂停一小段，避免请求过快被服务器断开
+            if idx % 100 == 0:
+                time.sleep(0.2)
+
+    # 3. 汇总
+    click.echo("")
+    summary = (
+        f"同步完成: {total} 只 | "
+        f"更新 {success} | 已是最新 {skipped} | "
+        f"失败 {failed} | 共写入 {total_written} 条"
+    )
+    click.echo(summary)

src/easy_tdx/offline/__init__.py

@@ -1,4 +1,4 @@
-"""离线数据读取模块 —— 从本地通达信安装目录读取数据文件。"""
+"""离线数据读写模块 —— 从本地通达信安装目录读取/写入数据文件。"""
 
 from .block import CustomerBlock, read_block_dat, read_customer_blocks
 from .daily_bar import find_daily_bar_file, read_daily_bars
@@ -8,21 +8,58 @@ from .gbbq import GbbqRecord, read_gbbq
 from .history_financial import read_history_financial
 from .min_bar import read_5min_bars, read_lc_min_bars
 from .paths import detect_tdx_home, resolve_vipdoc
+from .write_daily import (
+    append_daily_bars,
+    encode_daily_bar,
+    get_last_bar_date,
+    sync_daily_bars_from_security_bars,
+)
+from .write_ex_daily import (
+    append_ex_daily_bars,
+    encode_ex_daily_bar,
+    get_last_ex_bar_date,
+    sync_ex_daily_bars,
+)
+from .write_min_bar import (
+    append_5min_bars,
+    append_lc_min_bars,
+    encode_5min_bar,
+    encode_lc_min_bar,
+    get_last_5min_bar_datetime,
+    get_last_lc_min_bar_datetime,
+)
 
 __all__ = [
     # 路径
     "detect_tdx_home",
     "resolve_vipdoc",
-    # 日线
+    # 日线读取
     "read_daily_bars",
     "find_daily_bar_file",
-    # 分钟线
+    # 日线写入
+    "encode_daily_bar",
+    "append_daily_bars",
+    "get_last_bar_date",
+    "sync_daily_bars_from_security_bars",
+    # 扩展市场日线写入
+    "encode_ex_daily_bar",
+    "append_ex_daily_bars",
+    "get_last_ex_bar_date",
+    "sync_ex_daily_bars",
+    # 分钟线写入
+    "encode_5min_bar",
+    "encode_lc_min_bar",
+    "append_5min_bars",
+    "append_lc_min_bars",
+    "get_last_5min_bar_datetime",
+    "get_last_lc_min_bar_datetime",
+    # 分钟线读取
     "read_5min_bars",
     "read_lc_min_bars",
     "find_5min_bar_file",
     "find_lc1_bar_file",
     "find_lc5_bar_file",
-    # 扩展市场
+    # 扩展市场读取
     "ExDailyBar",
     "read_ex_daily_bars",
     # 板块

src/easy_tdx/offline/write_daily.py

@@ -0,0 +1,145 @@
+"""离线日线数据写入 —— 将 SecurityBar 编码并追加到 .day 文件。"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from ..models.bar import SecurityBar
+from .daily_bar import _DAILY_FMT
+
+__all__ = [
+    "encode_daily_bar",
+    "append_daily_bars",
+    "get_last_bar_date",
+    "sync_daily_bars_from_security_bars",
+]
+
+
+# ---------------------------------------------------------------------------
+# encode
+# ---------------------------------------------------------------------------
+
+
+def encode_daily_bar(
+    bar: SecurityBar,
+    price_coeff: float,
+    vol_coeff: float,
+) -> bytes:
+    """将 SecurityBar 编码为 32 字节 .day 记录。
+
+    Args:
+        bar: K 线数据（open/close/high/low 为实际价格，非整数）。
+        price_coeff: 价格系数（A 股 0.01，基金 0.001 等）。
+        vol_coeff: 成交量系数（A 股 0.01，指数 1.0 等）。
+
+    Returns:
+        32 字节的二进制记录。
+    """
+    date_int = bar.year * 10000 + bar.month * 100 + bar.day
+    return _DAILY_FMT.pack(
+        date_int,
+        int(round(bar.open / price_coeff)),
+        int(round(bar.high / price_coeff)),
+        int(round(bar.low / price_coeff)),
+        int(round(bar.close / price_coeff)),
+        bar.amount,  # float32, 由 struct 自动截断
+        int(round(bar.vol / vol_coeff)),
+        0,  # reserved
+    )
+
+
+# ---------------------------------------------------------------------------
+# query
+# ---------------------------------------------------------------------------
+
+
+def get_last_bar_date(filepath: str | Path) -> int | None:
+    """读取 .day 文件最后一条记录的日期。
+
+    Returns:
+        YYYYMMDD 整数，文件为空或太短时返回 None。
+    """
+    filepath = Path(filepath)
+    if not filepath.is_file():
+        return None
+    size = filepath.stat().st_size
+    if size < _DAILY_FMT.size:
+        return None
+    with filepath.open("rb") as f:
+        f.seek(size - _DAILY_FMT.size)
+        last_record = f.read(_DAILY_FMT.size)
+    (date_int, *_) = _DAILY_FMT.unpack(last_record)
+    return date_int
+
+
+def _bar_date_int(bar: SecurityBar) -> int:
+    return bar.year * 10000 + bar.month * 100 + bar.day
+
+
+# ---------------------------------------------------------------------------
+# append
+# ---------------------------------------------------------------------------
+
+
+def append_daily_bars(
+    filepath: str | Path,
+    bars: list[SecurityBar],
+    price_coeff: float,
+    vol_coeff: float,
+) -> int:
+    """将 bars 追加写入 .day 文件，自动跳过重复日期。
+
+    Args:
+        filepath: .day 文件路径。
+        bars: 待写入的 K 线列表（按时间升序）。
+        price_coeff: 价格系数。
+        vol_coeff: 成交量系数。
+
+    Returns:
+        实际写入的记录数。
+    """
+    filepath = Path(filepath)
+
+    # 获取文件末尾日期，用于去重
+    last_date = get_last_bar_date(filepath)
+
+    # 过滤出日期严格大于末尾的新记录
+    new_bars = (
+        [b for b in bars if _bar_date_int(b) > last_date] if last_date is not None else list(bars)
+    )
+
+    if not new_bars:
+        return 0
+
+    encoded = b"".join(encode_daily_bar(b, price_coeff, vol_coeff) for b in new_bars)
+    with filepath.open("ab") as f:
+        f.write(encoded)
+
+    return len(new_bars)
+
+
+# ---------------------------------------------------------------------------
+# sync
+# ---------------------------------------------------------------------------
+
+
+def sync_daily_bars_from_security_bars(
+    filepath: str | Path,
+    server_bars: list[SecurityBar],
+    price_coeff: float,
+    vol_coeff: float,
+) -> int:
+    """将服务端获取的日线数据同步写入本地 .day 文件。
+
+    完整流程：读取文件末尾日期 → 过滤新数据 → 追加写入。
+
+    Args:
+        filepath: .day 文件路径。
+        server_bars: 服务端返回的日线数据（按时间升序）。
+        price_coeff: 价格系数。
+        vol_coeff: 成交量系数。
+
+    Returns:
+        实际写入的记录数。
+    """
+    return append_daily_bars(filepath, server_bars, price_coeff, vol_coeff)

src/easy_tdx/offline/write_ex_daily.py

@@ -0,0 +1,93 @@
+"""离线扩展市场日线写入 —— 将 ExDailyBar 编码并追加到 .day 文件。"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from .ex_daily_bar import _EX_DAILY_FMT, ExDailyBar
+
+__all__ = [
+    "encode_ex_daily_bar",
+    "append_ex_daily_bars",
+    "get_last_ex_bar_date",
+    "sync_ex_daily_bars",
+]
+
+
+def encode_ex_daily_bar(bar: ExDailyBar) -> bytes:
+    """将 ExDailyBar 编码为 32 字节扩展市场 .day 记录。
+
+    扩展市场价格直接为 float32，无需系数转换。
+    """
+    date_int = bar.year * 10000 + bar.month * 100 + bar.day
+    return _EX_DAILY_FMT.pack(
+        date_int,
+        bar.open,
+        bar.high,
+        bar.low,
+        bar.close,
+        bar.amount,
+        bar.vol,
+        bar.settlement,
+    )
+
+
+def get_last_ex_bar_date(filepath: str | Path) -> int | None:
+    """读取扩展市场 .day 文件最后一条记录的日期。
+
+    Returns:
+        YYYYMMDD 整数，文件为空或太短时返回 None。
+    """
+    filepath = Path(filepath)
+    if not filepath.is_file():
+        return None
+    size = filepath.stat().st_size
+    if size < _EX_DAILY_FMT.size:
+        return None
+    with filepath.open("rb") as f:
+        f.seek(size - _EX_DAILY_FMT.size)
+        last_record = f.read(_EX_DAILY_FMT.size)
+    (date_int, *_) = _EX_DAILY_FMT.unpack(last_record)
+    return date_int
+
+
+def _bar_date_int(bar: ExDailyBar) -> int:
+    return bar.year * 10000 + bar.month * 100 + bar.day
+
+
+def append_ex_daily_bars(
+    filepath: str | Path,
+    bars: list[ExDailyBar],
+) -> int:
+    """将扩展市场 bars 追加写入 .day 文件，自动跳过重复日期。
+
+    Returns:
+        实际写入的记录数。
+    """
+    filepath = Path(filepath)
+    last_date = get_last_ex_bar_date(filepath)
+
+    new_bars = (
+        [b for b in bars if _bar_date_int(b) > last_date] if last_date is not None else list(bars)
+    )
+
+    if not new_bars:
+        return 0
+
+    encoded = b"".join(encode_ex_daily_bar(b) for b in new_bars)
+    with filepath.open("ab") as f:
+        f.write(encoded)
+
+    return len(new_bars)
+
+
+def sync_ex_daily_bars(
+    filepath: str | Path,
+    server_bars: list[ExDailyBar],
+) -> int:
+    """将服务端获取的扩展市场日线同步写入本地 .day 文件。
+
+    Returns:
+        实际写入的记录数。
+    """
+    return append_ex_daily_bars(filepath, server_bars)

src/easy_tdx/offline/write_min_bar.py

@@ -0,0 +1,186 @@
+"""离线分钟线写入 —— 将 SecurityBar 编码并追加到 .5 / .lc1 / .lc5 文件。"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from ..models.bar import SecurityBar
+from .min_bar import _LC_MIN_FMT, _MIN_FMT
+
+__all__ = [
+    "encode_5min_bar",
+    "encode_lc_min_bar",
+    "append_5min_bars",
+    "append_lc_min_bars",
+    "get_last_5min_bar_datetime",
+    "get_last_lc_min_bar_datetime",
+]
+
+
+# ---------------------------------------------------------------------------
+# date/time 编码（与 min_bar._decode_tdx_date/time 互逆）
+# ---------------------------------------------------------------------------
+
+
+def _encode_tdx_date(year: int, month: int, day: int) -> int:
+    """将 (year, month, day) 编码为 2 字节压缩日期。"""
+    return (year - 2004) * 2048 + month * 100 + day
+
+
+def _encode_tdx_time(hour: int, minute: int) -> int:
+    """将 (hour, minute) 编码为从 0:00 起的分钟数。"""
+    return hour * 60 + minute
+
+
+def _bar_datetime_key(bar: SecurityBar) -> tuple[int, int, int, int, int]:
+    return (bar.year, bar.month, bar.day, bar.hour, bar.minute)
+
+
+# ---------------------------------------------------------------------------
+# .5 文件 (OHLC 为整数 / 100)
+# ---------------------------------------------------------------------------
+
+
+def encode_5min_bar(bar: SecurityBar) -> bytes:
+    """将 SecurityBar 编码为 32 字节 .5 记录。
+
+    OHLC 为整数（实际价格 × 100），amount 为 float32。
+    """
+    return _MIN_FMT.pack(
+        _encode_tdx_date(bar.year, bar.month, bar.day),
+        _encode_tdx_time(bar.hour, bar.minute),
+        int(round(bar.open * 100)),
+        int(round(bar.high * 100)),
+        int(round(bar.low * 100)),
+        int(round(bar.close * 100)),
+        bar.amount,  # float32
+        int(round(bar.vol)),
+        0,  # reserved
+    )
+
+
+def get_last_5min_bar_datetime(
+    filepath: str | Path,
+) -> tuple[int, int, int, int, int] | None:
+    """读取 .5 文件最后一条记录的日期时间。
+
+    Returns:
+        (year, month, day, hour, minute) 元组，文件为空时返回 None。
+    """
+    filepath = Path(filepath)
+    if not filepath.is_file():
+        return None
+    size = filepath.stat().st_size
+    if size < _MIN_FMT.size:
+        return None
+    with filepath.open("rb") as f:
+        f.seek(size - _MIN_FMT.size)
+        last_record = f.read(_MIN_FMT.size)
+    from .min_bar import _decode_tdx_date, _decode_tdx_time
+
+    date_num, time_num, *_ = _MIN_FMT.unpack(last_record)
+    year, month, day = _decode_tdx_date(date_num)
+    hour, minute = _decode_tdx_time(time_num)
+    return (year, month, day, hour, minute)
+
+
+def append_5min_bars(
+    filepath: str | Path,
+    bars: list[SecurityBar],
+) -> int:
+    """将 5 分钟线 bars 追加写入 .5 文件，自动跳过重复时间点。
+
+    Returns:
+        实际写入的记录数。
+    """
+    filepath = Path(filepath)
+    last_dt = get_last_5min_bar_datetime(filepath)
+
+    if last_dt is not None:
+        new_bars = [b for b in bars if _bar_datetime_key(b) > last_dt]
+    else:
+        new_bars = list(bars)
+
+    if not new_bars:
+        return 0
+
+    encoded = b"".join(encode_5min_bar(b) for b in new_bars)
+    with filepath.open("ab") as f:
+        f.write(encoded)
+
+    return len(new_bars)
+
+
+# ---------------------------------------------------------------------------
+# .lc1 / .lc5 文件 (OHLC 为 float32)
+# ---------------------------------------------------------------------------
+
+
+def encode_lc_min_bar(bar: SecurityBar) -> bytes:
+    """将 SecurityBar 编码为 32 字节 .lc1/.lc5 记录。
+
+    OHLC 为 float32，无需系数转换。
+    """
+    return _LC_MIN_FMT.pack(
+        _encode_tdx_date(bar.year, bar.month, bar.day),
+        _encode_tdx_time(bar.hour, bar.minute),
+        bar.open,
+        bar.high,
+        bar.low,
+        bar.close,
+        bar.amount,  # float32
+        int(round(bar.vol)),
+        0,  # reserved
+    )
+
+
+def get_last_lc_min_bar_datetime(
+    filepath: str | Path,
+) -> tuple[int, int, int, int, int] | None:
+    """读取 .lc1/.lc5 文件最后一条记录的日期时间。
+
+    Returns:
+        (year, month, day, hour, minute) 元组，文件为空时返回 None。
+    """
+    filepath = Path(filepath)
+    if not filepath.is_file():
+        return None
+    size = filepath.stat().st_size
+    if size < _LC_MIN_FMT.size:
+        return None
+    with filepath.open("rb") as f:
+        f.seek(size - _LC_MIN_FMT.size)
+        last_record = f.read(_LC_MIN_FMT.size)
+    from .min_bar import _decode_tdx_date, _decode_tdx_time
+
+    date_num, time_num, *_ = _LC_MIN_FMT.unpack(last_record)
+    year, month, day = _decode_tdx_date(date_num)
+    hour, minute = _decode_tdx_time(time_num)
+    return (year, month, day, hour, minute)
+
+
+def append_lc_min_bars(
+    filepath: str | Path,
+    bars: list[SecurityBar],
+) -> int:
+    """将分钟线 bars 追加写入 .lc1/.lc5 文件，自动跳过重复时间点。
+
+    Returns:
+        实际写入的记录数。
+    """
+    filepath = Path(filepath)
+    last_dt = get_last_lc_min_bar_datetime(filepath)
+
+    if last_dt is not None:
+        new_bars = [b for b in bars if _bar_datetime_key(b) > last_dt]
+    else:
+        new_bars = list(bars)
+
+    if not new_bars:
+        return 0
+
+    encoded = b"".join(encode_lc_min_bar(b) for b in new_bars)
+    with filepath.open("ab") as f:
+        f.write(encoded)
+
+    return len(new_bars)