wbt 0.4.2

//! pairs 聚合去重 + 每年最赚/最亏关键交易。
//!
//! 业务口径（见飞书设计文档 §九 / §十一）：
//! - **聚合键**：`(symbol, 开仓时间, 平仓时间)`。引擎按 LIFO 撮合，一次平仓可能命中多个
//!   lot，从而产出多条共享同一 `(open_dt, close_dt)` 的部分成交记录；按此键聚合即去重。
//! - **count**：被合并的原始开平记录数量。
//! - **盈亏比例**：按成交量（`counts`）加权平均的 `profit_bp`，单位 BP（与 `pairs` 的
//!   `盈亏比例` 列一致）。
//! - **持仓K线数**：同组共享同一值（由开/平 bar 决定），取其值，**不求和**。
//! - **分年依据**：按**平仓时间**的年份。

use crate::core::errors::WbtError;
use crate::core::native_engine::{PairsSoA, dt_to_date_key_fast};
use polars::prelude::*;
use std::collections::hash_map::Entry;
use std::collections::{BTreeMap, HashMap};

/// 一条聚合后的开平记录。
#[derive(Debug, Clone, PartialEq)]
pub struct AggRow {
    pub sym_id: u32,
    pub open_dt: i64,
    pub close_dt: i64,
    pub dir: &'static str,
    pub open_price: f64,
    pub close_price: f64,
    pub hold_bars: i64,
    /// 成交量加权平均盈亏比例（BP）
    pub profit_bp: f64,
    /// 被合并的原始记录数量
    pub count: i64,
}

/// 按 `(sym_id, open_dt, close_dt)` 聚合 `PairsSoA`，结果按该键升序稳定排列。
pub fn aggregate_pairs(pairs: &PairsSoA) -> Vec<AggRow> {
    // 累加器：(profit*vol 之和, vol 之和, 记录数, hold_bars, dir, open_px, close_px)
    struct Acc {
        sum_pw: f64,
        sum_w: f64,
        count: i64,
        hold_bars: i64,
        dir: &'static str,
        open_price: f64,
        close_price: f64,
    }

    let mut groups: BTreeMap<(u32, i64, i64), Acc> = BTreeMap::new();

    for i in 0..pairs.sym_ids.len() {
        let key = (pairs.sym_ids[i], pairs.open_dts[i], pairs.close_dts[i]);
        let w = pairs.counts[i] as f64;
        let entry = groups.entry(key).or_insert_with(|| Acc {
            sum_pw: 0.0,
            sum_w: 0.0,
            count: 0,
            hold_bars: pairs.hold_bars[i],
            dir: pairs.dirs[i],
            open_price: pairs.open_prices[i],
            close_price: pairs.close_prices[i],
        });
        entry.sum_pw += pairs.profit_bps[i] * w;
        entry.sum_w += w;
        entry.count += 1;
        // 同组 hold_bars 理论一致；取较大值以防御异常输入。
        if pairs.hold_bars[i] > entry.hold_bars {
            entry.hold_bars = pairs.hold_bars[i];
        }
    }

    groups
        .into_iter()
        .map(|((sym_id, open_dt, close_dt), a)| {
            let profit_bp = if a.sum_w != 0.0 {
                // 与引擎 profit_bp 一致，保留两位小数
                (a.sum_pw / a.sum_w * 100.0).round() / 100.0
            } else {
                0.0
            };
            AggRow {
                sym_id,
                open_dt,
                close_dt,
                dir: a.dir,
                open_price: a.open_price,
                close_price: a.close_price,
                hold_bars: a.hold_bars,
                profit_bp,
                count: a.count,
            }
        })
        .collect()
}

/// 在聚合结果上按平仓年份分组，取每年 best/worst 各 `top` 笔。
///
/// 返回 `(year, kind, &AggRow)`，`kind` ∈ {"best", "worst"}；按 `(year, kind, 排名)` 顺序排列。
fn select_key_trades(
    agg: &[AggRow],
    top: usize,
    time_unit: TimeUnit,
) -> Vec<(i32, &'static str, usize)> {
    // year -> agg 行下标列表
    let mut by_year: BTreeMap<i32, Vec<usize>> = BTreeMap::new();
    for (idx, row) in agg.iter().enumerate() {
        let year = dt_to_date_key_fast(row.close_dt, time_unit) / 10000;
        by_year.entry(year).or_default().push(idx);
    }

    let mut out: Vec<(i32, &'static str, usize)> = Vec::new();
    for (year, mut idxs) in by_year {
        // best：profit 降序；tiebreak open_dt 升序保证稳定
        idxs.sort_by(|&a, &b| {
            agg[b]
                .profit_bp
                .total_cmp(&agg[a].profit_bp)
                .then(agg[a].open_dt.cmp(&agg[b].open_dt))
        });
        let mut best: Vec<usize> = Vec::new();
        for &i in idxs.iter().take(top) {
            best.push(i);
            out.push((year, "best", i));
        }
        // worst：profit 升序；剔除已入 best 的交易，避免同一笔同时是最赚和最亏
        idxs.sort_by(|&a, &b| {
            agg[a]
                .profit_bp
                .total_cmp(&agg[b].profit_bp)
                .then(agg[a].open_dt.cmp(&agg[b].open_dt))
        });
        for &i in idxs.iter().filter(|i| !best.contains(i)).take(top) {
            out.push((year, "worst", i));
        }
    }
    out
}

fn agg_rows_to_columns(
    pairs: &PairsSoA,
    rows: impl Iterator<Item = (Option<(i32, &'static str)>, AggRow)>,
) -> Result<DataFrame, WbtError> {
    let mut years: Vec<i32> = Vec::new();
    let mut kinds: Vec<&'static str> = Vec::new();
    let mut symbols: Vec<String> = Vec::new();
    let mut open_dts: Vec<i64> = Vec::new();
    let mut close_dts: Vec<i64> = Vec::new();
    let mut dirs: Vec<&'static str> = Vec::new();
    let mut open_px: Vec<f64> = Vec::new();
    let mut close_px: Vec<f64> = Vec::new();
    let mut hold_bars: Vec<i64> = Vec::new();
    let mut profits: Vec<f64> = Vec::new();
    let mut counts: Vec<i64> = Vec::new();

    let mut has_year = false;
    for (meta, r) in rows {
        if let Some((y, kind)) = meta {
            has_year = true;
            years.push(y);
            kinds.push(kind);
        }
        symbols.push(pairs.symbol_dict[r.sym_id as usize].clone());
        open_dts.push(r.open_dt);
        close_dts.push(r.close_dt);
        dirs.push(r.dir);
        open_px.push(r.open_price);
        close_px.push(r.close_price);
        hold_bars.push(r.hold_bars);
        profits.push(r.profit_bp);
        counts.push(r.count);
    }

    let open_series = Series::new("开仓时间".into(), &open_dts)
        .cast(&DataType::Datetime(pairs.time_unit, None))
        .map_err(WbtError::Polars)?;
    let close_series = Series::new("平仓时间".into(), &close_dts)
        .cast(&DataType::Datetime(pairs.time_unit, None))
        .map_err(WbtError::Polars)?;

    let mut cols: Vec<Column> = Vec::new();
    if has_year {
        cols.push(Series::new("year".into(), &years).into_column());
        cols.push(Series::new("kind".into(), &kinds).into_column());
    }
    cols.push(Series::new("symbol".into(), &symbols).into_column());
    cols.push(Series::new("交易方向".into(), &dirs).into_column());
    cols.push(open_series.into_column());
    cols.push(close_series.into_column());
    cols.push(Series::new("开仓价格".into(), &open_px).into_column());
    cols.push(Series::new("平仓价格".into(), &close_px).into_column());
    cols.push(Series::new("持仓K线数".into(), &hold_bars).into_column());
    cols.push(Series::new("盈亏比例".into(), &profits).into_column());
    cols.push(Series::new("count".into(), &counts).into_column());

    DataFrame::new_infer_height(cols).map_err(WbtError::Polars)
}

/// 由已聚合的 `AggRow` 切片构建聚合开平记录表（供 WeightBacktest 缓存复用）。
pub fn agg_to_df(pairs: &PairsSoA, agg: &[AggRow]) -> Result<DataFrame, WbtError> {
    agg_rows_to_columns(pairs, agg.iter().map(|r| (None, r.clone())))
}

/// 由已聚合的 `AggRow` 切片选取每年 best/worst 各 `top` 笔（含 `year` / `kind` 列）。
pub fn key_trades_to_df(
    pairs: &PairsSoA,
    agg: &[AggRow],
    top: usize,
) -> Result<DataFrame, WbtError> {
    // 选榜前先把 LIFO 拆分的同一段持仓还原为真实持仓段（按 open_dt 或 close_dt 任一相同
    // 传递合并），避免同段被重复枚举、重复上榜。aggregated_pairs 仍用未合并的 agg，语义不变。
    let merged = merge_holding_segments(agg);
    let selected = select_key_trades(&merged, top, pairs.time_unit);
    agg_rows_to_columns(
        pairs,
        selected
            .into_iter()
            .map(|(y, kind, i)| (Some((y, kind)), merged[i].clone())),
    )
}

/// 聚合去重后的开平记录表（聚合 + 物化，便于单测/单次调用）。
pub fn aggregated_pairs_df(pairs: &PairsSoA) -> Result<DataFrame, WbtError> {
    agg_to_df(pairs, &aggregate_pairs(pairs))
}

/// 每年最赚/最亏各 `top` 笔关键交易（聚合 + 物化，便于单测/单次调用）。
pub fn key_trades_df(pairs: &PairsSoA, top: usize) -> Result<DataFrame, WbtError> {
    key_trades_to_df(pairs, &aggregate_pairs(pairs), top)
}

/// 并查集 find（路径压缩）。
fn uf_find(parent: &mut [usize], mut x: usize) -> usize {
    while parent[x] != x {
        parent[x] = parent[parent[x]];
        x = parent[x];
    }
    x
}

/// 并查集 union。
fn uf_union(parent: &mut [usize], a: usize, b: usize) {
    let (ra, rb) = (uf_find(parent, a), uf_find(parent, b));
    if ra != rb {
        parent[ra] = rb;
    }
}

/// 把 `aggregate_pairs` 的结果按真实持仓段进一步合并，用于 `key_trades` 选榜。
///
/// `aggregate_pairs` 仅按 `(symbol, open_dt, close_dt)` 精确聚合，无法合并 LIFO 撮合
/// 对同一段持仓的拆分：一次开仓分批平仓（同 `open_dt`、不同 `close_dt`）、分批开仓一次
/// 平仓（不同 `open_dt`、同 `close_dt`）。本函数按 `(sym_id, dir)` 分组，对 `open_dt`
/// 或 `close_dt` 任一相同的记录用并查集做传递合并（覆盖 A-B 同开、B-C 同平的链式合并）；
/// 每个连通分量取持仓 K 线数最长的一笔为代表，`count` 取成员之和；`profit_bp` /
/// `open_price` / `close_price` 等价格类字段直接取代表笔——同一连通分量的成员之间
/// 至多共享 open 或 close 中的一端，价格未必相等；选榜语义下「持仓最长的那段」即
/// 「最具代表性」的主升段，足以代表整段持仓的展示口径，故不再按价格重算。
///
/// 不变量：输出中同 `(sym_id, dir)` 的 `open_dt`、`close_dt` 各自唯一——任意两条同
/// `open_dt`（或同 `close_dt`）必被并入同一分量，不会落到两个代表笔上。
pub fn merge_holding_segments(agg: &[AggRow]) -> Vec<AggRow> {
    let n = agg.len();
    if n == 0 {
        return Vec::new();
    }

    let mut parent: Vec<usize> = (0..n).collect();
    // 用 (sym_id, dir, dt) 作映射键，天然隔离不同 (sym, dir) 组。
    let mut open_seen: HashMap<(u32, &'static str, i64), usize> = HashMap::with_capacity(n);
    let mut close_seen: HashMap<(u32, &'static str, i64), usize> = HashMap::with_capacity(n);
    for (i, r) in agg.iter().enumerate() {
        match open_seen.entry((r.sym_id, r.dir, r.open_dt)) {
            Entry::Occupied(e) => uf_union(&mut parent, i, *e.get()),
            Entry::Vacant(e) => {
                e.insert(i);
            }
        }
        match close_seen.entry((r.sym_id, r.dir, r.close_dt)) {
            Entry::Occupied(e) => uf_union(&mut parent, i, *e.get()),
            Entry::Vacant(e) => {
                e.insert(i);
            }
        }
    }

    // 按连通分量根聚合：选代表笔（hold_bars 最长，平局取 open_dt 更早以稳定）+ 累加 count。
    let mut rep: HashMap<usize, usize> = HashMap::new();
    let mut count_sum: HashMap<usize, i64> = HashMap::new();
    for (i, r) in agg.iter().enumerate() {
        let root = uf_find(&mut parent, i);
        *count_sum.entry(root).or_insert(0) += r.count;
        match rep.get(&root) {
            None => {
                rep.insert(root, i);
            }
            Some(&cur) => {
                let better = r.hold_bars > agg[cur].hold_bars
                    || (r.hold_bars == agg[cur].hold_bars && r.open_dt < agg[cur].open_dt);
                if better {
                    rep.insert(root, i);
                }
            }
        }
    }

    let mut out: Vec<AggRow> = rep
        .iter()
        .map(|(&root, &idx)| {
            let mut row = agg[idx].clone();
            row.count = count_sum[&root];
            row
        })
        .collect();

    // 稳定输出顺序，便于下游与测试。
    out.sort_by(|a, b| {
        a.sym_id
            .cmp(&b.sym_id)
            .then_with(|| a.dir.cmp(b.dir))
            .then_with(|| a.open_dt.cmp(&b.open_dt))
            .then_with(|| a.close_dt.cmp(&b.close_dt))
    });
    out
}

#[cfg(test)]
mod tests {
    use super::*;

    /// 构造一个最小 PairsSoA。每条记录用 (sym_id, open_dt, close_dt, hold_bars, profit_bp, count)。
    fn make_pairs(rows: &[(u32, i64, i64, i64, f64, i64)], symbol_dict: Vec<String>) -> PairsSoA {
        PairsSoA {
            sym_ids: rows.iter().map(|r| r.0).collect(),
            dirs: rows.iter().map(|_| "多头").collect(),
            open_dts: rows.iter().map(|r| r.1).collect(),
            close_dts: rows.iter().map(|r| r.2).collect(),
            open_prices: rows.iter().map(|_| 100.0).collect(),
            close_prices: rows.iter().map(|_| 110.0).collect(),
            hold_bars: rows.iter().map(|r| r.3).collect(),
            event_seqs: rows.iter().map(|_| "开多->平多").collect(),
            profit_bps: rows.iter().map(|r| r.4).collect(),
            counts: rows.iter().map(|r| r.5).collect(),
            time_unit: TimeUnit::Milliseconds,
            symbol_dict,
        }
    }

    // 2024-06-03 与 2025-06-03 的毫秒时间戳
    const DT_2024: i64 = 1_717_372_800_000; // 2024-06-03 00:00:00 UTC
    const DT_2024B: i64 = 1_717_459_200_000; // 2024-06-04
    const DT_2025: i64 = 1_748_908_800_000; // 2025-06-03 00:00:00 UTC

    #[test]
    fn aggregate_empty_is_empty() {
        let pairs = make_pairs(&[], vec!["A".into()]);
        assert!(aggregate_pairs(&pairs).is_empty());
        let df = aggregated_pairs_df(&pairs).unwrap();
        assert_eq!(df.height(), 0);
        // schema 仍齐全
        assert!(df.column("count").is_ok());
        assert!(df.column("盈亏比例").is_ok());
    }

    #[test]
    fn aggregate_merges_same_open_close() {
        // 同一 (sym, open, close) 的两条部分成交：profit 100/200，量 1/3
        let pairs = make_pairs(
            &[
                (0, DT_2024, DT_2024B, 5, 100.0, 1),
                (0, DT_2024, DT_2024B, 5, 200.0, 3),
            ],
            vec!["A".into()],
        );
        let agg = aggregate_pairs(&pairs);
        assert_eq!(agg.len(), 1, "同 key 应合并为一条");
        let r = &agg[0];
        assert_eq!(r.count, 2, "count = 合并的记录数");
        assert_eq!(r.hold_bars, 5, "hold_bars 取共享值，不求和");
        // 量加权平均 = (100*1 + 200*3) / (1+3) = 175
        assert!((r.profit_bp - 175.0).abs() < 1e-9, "got {}", r.profit_bp);
    }

    #[test]
    fn aggregate_keeps_distinct_close() {
        let pairs = make_pairs(
            &[
                (0, DT_2024, DT_2024B, 5, 100.0, 1),
                (0, DT_2024, DT_2025, 5, 200.0, 1),
            ],
            vec!["A".into()],
        );
        assert_eq!(aggregate_pairs(&pairs).len(), 2, "平仓时间不同应保留两条");
    }

    #[test]
    fn key_trades_best_worst_per_year() {
        // 2024 年 3 笔：profit 300/-50/10；top=2
        let pairs = make_pairs(
            &[
                (0, DT_2024, DT_2024B, 1, 300.0, 1),
                (1, DT_2024, DT_2024B, 1, -50.0, 1),
                (2, DT_2024, DT_2024B, 1, 10.0, 1),
            ],
            vec!["A".into(), "B".into(), "C".into()],
        );
        let agg = aggregate_pairs(&pairs);
        let sel = select_key_trades(&agg, 2, TimeUnit::Milliseconds);
        let best: Vec<f64> = sel
            .iter()
            .filter(|(_, k, _)| *k == "best")
            .map(|(_, _, i)| agg[*i].profit_bp)
            .collect();
        let worst: Vec<f64> = sel
            .iter()
            .filter(|(_, k, _)| *k == "worst")
            .map(|(_, _, i)| agg[*i].profit_bp)
            .collect();
        assert_eq!(best, vec![300.0, 10.0], "best 降序取前 2");
        // worst 升序取前 2，但 10.0 已入 best 被剔除 → 仅剩 -50.0（去重避免同笔既最赚又最亏）
        assert_eq!(worst, vec![-50.0], "worst 剔除已入 best 的交易");
    }

    #[test]
    fn key_trades_top_exceeds_count() {
        // 当年仅 2 笔，top=3 → 各取 2 笔，不补空、不 panic
        let pairs = make_pairs(
            &[
                (0, DT_2024, DT_2024B, 1, 100.0, 1),
                (1, DT_2024, DT_2024B, 1, -20.0, 1),
            ],
            vec!["A".into(), "B".into()],
        );
        let agg = aggregate_pairs(&pairs);
        let sel = select_key_trades(&agg, 3, TimeUnit::Milliseconds);
        // 2 笔均入 best；worst 剔除 best 后为空 → 不重复、不补空、不 panic
        assert_eq!(sel.iter().filter(|(_, k, _)| *k == "best").count(), 2);
        assert_eq!(sel.iter().filter(|(_, k, _)| *k == "worst").count(), 0);
    }

    #[test]
    fn key_trades_groups_by_close_year() {
        let pairs = make_pairs(
            &[
                (0, DT_2024, DT_2024B, 1, 100.0, 1),
                (1, DT_2024, DT_2025, 1, 200.0, 1),
            ],
            vec!["A".into(), "B".into()],
        );
        let agg = aggregate_pairs(&pairs);
        let sel = select_key_trades(&agg, 5, TimeUnit::Milliseconds);
        let years: std::collections::BTreeSet<i32> = sel.iter().map(|(y, _, _)| *y).collect();
        assert_eq!(years, [2024, 2025].into_iter().collect(), "按平仓年份分组");
    }

    #[test]
    fn key_trades_df_has_year_kind_columns() {
        let pairs = make_pairs(&[(0, DT_2024, DT_2024B, 1, 100.0, 1)], vec!["A".into()]);
        let df = key_trades_df(&pairs, 3).unwrap();
        assert!(df.column("year").is_ok());
        assert!(df.column("kind").is_ok());
        assert!(df.column("symbol").is_ok());
        assert_eq!(df.height(), 1, "1 笔 → 仅 best 1（worst 去重后为空）");
    }

    // ---- merge_holding_segments：真实持仓段合并（任务 t101322 新增） ----

    /// 构造 AggRow 切片。每条用 (sym_id, dir, open_dt, close_dt, hold_bars, profit_bp, count)。
    fn make_agg(rows: &[(u32, &'static str, i64, i64, i64, f64, i64)]) -> Vec<AggRow> {
        rows.iter()
            .map(
                |&(sym_id, dir, open_dt, close_dt, hold_bars, profit_bp, count)| AggRow {
                    sym_id,
                    open_dt,
                    close_dt,
                    dir,
                    open_price: 100.0,
                    close_price: 110.0,
                    hold_bars,
                    profit_bp,
                    count,
                },
            )
            .collect()
    }

    /// 同 open、分批平仓（同 open 不同 close）应合并为一段：代表取持仓最长笔，count 求和。
    #[test]
    fn merge_same_open_split_close() {
        let agg = make_agg(&[
            (0, "多头", 100, 200, 5, 10.0, 2),
            (0, "多头", 100, 300, 8, 20.0, 3),
        ]);
        let m = merge_holding_segments(&agg);
        assert_eq!(m.len(), 1, "同 open 的两条应合并为一段");
        assert_eq!(m[0].hold_bars, 8, "代表取持仓 K 线数最长者");
        assert_eq!(m[0].close_dt, 300);
        assert_eq!(m[0].profit_bp, 20.0, "profit 取代表笔");
        assert_eq!(m[0].count, 5, "count 取成员之和 2+3");
    }

    /// 传递合并：A-B 同 open、B-C 同 close 应并成一段。
    #[test]
    fn merge_transitive() {
        let agg = make_agg(&[
            (0, "多头", 100, 200, 5, 10.0, 1),
            (0, "多头", 100, 400, 9, 99.0, 2),
            (0, "多头", 300, 400, 4, 30.0, 3),
        ]);
        let m = merge_holding_segments(&agg);
        assert_eq!(m.len(), 1, "A-B 同开、B-C 同平应链式合并");
        assert_eq!(m[0].hold_bars, 9);
        assert_eq!(m[0].open_dt, 100);
        assert_eq!(m[0].close_dt, 400);
        assert_eq!(m[0].count, 6, "1+2+3");
    }

    /// 独立段（open、close 均不同）不应合并。
    #[test]
    fn merge_independent_not_merged() {
        let agg = make_agg(&[
            (0, "多头", 100, 150, 5, 10.0, 1),
            (0, "多头", 200, 250, 5, 20.0, 1),
        ]);
        assert_eq!(merge_holding_segments(&agg).len(), 2);
    }

    /// 同 open、同 close 但不同 symbol 或不同方向，不应跨组合并。
    #[test]
    fn merge_sym_and_dir_isolation() {
        let by_sym = make_agg(&[
            (0, "多头", 100, 200, 5, 10.0, 1),
            (1, "多头", 100, 200, 5, 20.0, 1),
        ]);
        assert_eq!(
            merge_holding_segments(&by_sym).len(),
            2,
            "不同 symbol 不合并"
        );
        let by_dir = make_agg(&[
            (0, "多头", 100, 200, 5, 10.0, 1),
            (0, "空头", 100, 200, 5, 20.0, 1),
        ]);
        assert_eq!(merge_holding_segments(&by_dir).len(), 2, "不同方向不合并");
    }

    /// 不变量：输出中同 (sym, dir) 的 open_dt、close_dt 各自唯一。
    #[test]
    fn merge_invariant_unique_open_close() {
        let agg = make_agg(&[
            (0, "多头", 100, 200, 5, 10.0, 1),
            (0, "多头", 100, 400, 9, 99.0, 2),
            (0, "多头", 300, 400, 4, 30.0, 3),
            (0, "空头", 500, 600, 3, -5.0, 4),
            (1, "多头", 100, 200, 2, 7.0, 5),
        ]);
        let m = merge_holding_segments(&agg);
        use std::collections::HashSet;
        let mut opens: HashSet<(u32, &str, i64)> = HashSet::new();
        let mut closes: HashSet<(u32, &str, i64)> = HashSet::new();
        for s in &m {
            assert!(opens.insert((s.sym_id, s.dir, s.open_dt)), "open_dt 重复");
            assert!(
                closes.insert((s.sym_id, s.dir, s.close_dt)),
                "close_dt 重复"
            );
        }
    }

    #[test]
    fn merge_empty() {
        assert!(merge_holding_segments(&[]).is_empty());
    }

    /// 端到端：LIFO 拆成多条 AggRow 的同一段持仓，key_trades 去重后只上榜一次。
    #[test]
    fn key_trades_dedups_lifo_split() {
        // 同 sym 同 open，分两次平仓（close 不同但同年）→ aggregate 后 2 条，merge 后 1 段。
        let pairs = make_pairs(
            &[
                (0, DT_2024, DT_2024, 5, 100.0, 1),
                (0, DT_2024, DT_2024B, 8, 300.0, 1),
            ],
            vec!["A".into()],
        );
        assert_eq!(
            aggregate_pairs(&pairs).len(),
            2,
            "aggregate 按 close 区分仍是 2 条"
        );
        let df = key_trades_df(&pairs, 3).unwrap();
        assert_eq!(df.height(), 1, "LIFO 拆分的同段去重后只上榜一次");
        assert_eq!(
            df.column("count").unwrap().i64().unwrap().get(0),
            Some(2),
            "count=1+1"
        );
        assert_eq!(
            df.column("盈亏比例").unwrap().f64().unwrap().get(0),
            Some(300.0),
            "profit 取代表笔（持仓最长）"
        );
    }
}