2026年重磅升级已全面落地！欢迎加入专注财经数据与量化投研的【数据科学实战】知识星球！您将获取持续更新的《财经数据宝典》与《量化投研宝典》，双典协同提供系统化指引；星球内含300篇以上独有高质量文章，深度覆盖策略开发、因子分析、风险管理等核心领域，内容基本每日更新；同步推出的「量化因子专题教程」系列（含完整可运行代码与实战案例），系统详解因子构建、回测与优化全流程，并实现日更迭代。我们持续扩充独家内容资源，全方位赋能您的投研效率与专业成长。无论您是量化新手还是资深研究者，这里都是助您少走弯路、事半功倍的理想伙伴，携手共探数据驱动的投资未来！

引言

你是否厌倦了传统的 K 线图分析？是否想要一种能够过滤市场噪音、专注于价格趋势的交易方式？今天我要介绍的 Kagi 图表交易系统，可能会给你带来全新的视角。

Kagi 图表是一种源自日本的图表类型，它与传统的 K 线图最大的不同在于：完全忽略时间维度，只关注价格变动。只有当价格反转超过特定幅度时，Kagi 线才会改变方向。这使得它在趋势跟踪方面具有独特优势。

更重要的是，本文介绍的系统不是简单的单仓位策略，而是一个能够同时管理 20 到 50 个独立仓位的完整交易生态系统。每个仓位都有自己的止损和止盈，就像一个迷你的投资组合管理器。

本文将带你从零开始，用 Python 构建一个完整的多仓位 Kagi 回测系统。

核心概念：什么是多仓位 Kagi 交易

传统交易系统通常一次只开一个仓位：开仓、等待平仓、再开新仓。但市场往往同时给出多个交易机会，为什么不同时捕捉它们呢？

核心思路：Kagi 图表在不同的市场结构点生成信号。有些信号出现在强趋势中，有些出现在反转点。我们不必只选择一个入场点，而是可以开设多个仓位，让每个仓位按照自己的规则独立运行。

系统工作流程如下：

市场价格变动    ↓Kagi 图表构建（过滤噪音）    ↓信号生成（阴线转阳线、阳线转阴线）    ↓开仓（最多可同时持有 N 个仓位）    ↓独立管理（每笔交易有自己的止损/止盈）    ↓达到止损/止盈或超过最大持仓时间时平仓

Kagi 状态机的 Python 实现

Kagi 图表的核心是一个状态机。它要么处于上升趋势，要么处于下降趋势；线型要么是阳线（粗线/强势），要么是阴线（细线/弱势）。

from dataclasses import dataclassfrom typing import Optionalfrom datetime import datetime@dataclassclass KagiState:    """Kagi 图表状态类，记录当前市场结构"""    reference_price: float = 0.0       # 参考价格（当前基准点）    reference_time: Optional[datetime] = None  # 参考时间    local_minimum: float = float('inf')  # 近期最低点（用于止损计算）    local_maximum: float = float('-inf') # 近期最高点（用于止损计算）    is_uptrend: bool = True            # 是否处于上升趋势    is_downtrend: bool = False         # 是否处于下降趋势    is_yang: bool = True               # 是否为阳线（粗线，强趋势）    is_yin: bool = False               # 是否为阴线（细线，弱趋势）    last_reference_price: float = 0.0  # 上一个参考价格

这四个布尔状态（上升/下降趋势、阳线/阴线）驱动所有的交易决策。

仓位数据结构设计

每笔交易都需要自己的身份标识和管理参数。与简单系统只跟踪一个订单号不同，我们需要为每个仓位维护完整的档案：

@dataclassclass Position:    """单个仓位的完整信息"""    position_id: int           # 仓位唯一标识    direction: str             # 方向：'long' 或 'short'    entry_price: float         # 入场价格    entry_time: datetime       # 入场时间    size: float                # 仓位大小    stop_loss: float           # 止损价格    take_profit: float         # 止盈价格    is_active: bool = True     # 仓位是否活跃    trailing_level_1: Optional[float] = None  # 移动止损激活价格    trailing_stop_1: Optional[float] = None   # 移动止损价格    trailing_1_active: bool = False           # 移动止损是否已激活@dataclassclass Trade:    """已完成的交易记录"""    position_id: int           # 仓位 ID    entry_time: datetime       # 入场时间    exit_time: datetime        # 出场时间    direction: str             # 交易方向    entry_price: float         # 入场价格    exit_price: float          # 出场价格    size: float                # 仓位大小    pnl: float                 # 盈亏金额    pnl_percent: float         # 盈亏百分比    exit_reason: str           # 出场原因（止损/止盈/超时等）    bars_held: int = 0         # 持仓 K 线数量

这种设计允许独立管理每个仓位——一个仓位可能正在追踪移动止损，而另一个仓位还在等待激活。

多仓位回测器核心实现

下面是回测系统的主要框架：

from enum import Enumfrom typing import List, Tupleimport pandas as pdclass SignalMode(Enum):    """信号生成模式"""    YIN_YANG_ONLY = "yin_yang_only"        # 仅阴阳转换信号    TREND_REVERSAL = "trend_reversal"      # 仅趋势反转信号    AGGRESSIVE = "aggressive"               # 激进模式    ULTRA_AGGRESSIVE = "ultra_aggressive"   # 超激进模式    EVERY_KAGI_LINE = "every_kagi_line"    # 每条新 Kagi 线    EVERY_BAR = "every_bar"                # 每根 K 线都产生信号class MultiPositionKagiBacktester:    """    多仓位 Kagi 回测器    特点：    - 支持多个仓位同时持有    - 每个仓位独立管理止损止盈    - 新仓位不会影响现有仓位    """    def __init__(        self,        reversal_amount: float = 0.03,         # 反转阈值（百分比）        max_concurrent_positions: int = 20,     # 最大同时持仓数        risk_per_trade_percent: float = 0.3,   # 每笔交易风险百分比        risk_reward_ratio: float = 1.5,        # 风险回报比        signal_mode: SignalMode = SignalMode.ULTRA_AGGRESSIVE,        stop_loss_buffer: float = 3.0,         # 止损缓冲百分比        initial_capital: float = 10000.0       # 初始资金):        self.reversal_amount = reversal_amount        self.max_concurrent_positions = max_concurrent_positions        self.risk_per_trade_percent = risk_per_trade_percent        self.risk_reward_ratio = risk_reward_ratio        self.signal_mode = signal_mode        self.stop_loss_buffer = stop_loss_buffer        self.initial_capital = initial_capital        self.current_capital = initial_capital        # 状态追踪        self.kagi_state = KagiState()        self.active_positions: List[Position] = []  # 活跃仓位列表        self.trades: List[Trade] = []               # 已完成交易列表        self.equity_curve: List[Tuple[datetime, float]] = []  # 权益曲线        self.next_position_id = 1        self.bars_since_last_signal = 0

信号生成逻辑

系统能够检测六种不同类型的 Kagi 转换，每种都代表潜在的交易机会：

def update_kagi_chart(self, bar: pd.Series) -> Optional[str]:    """    更新 Kagi 图表并生成交易信号    返回值：        'buy' - 买入信号        'sell' - 卖出信号        None - 无信号    """    current_close = bar['close']    current_time = bar.name    # 计算反转阈值    reversal_amt = self.kagi_state.reference_price * (self.reversal_amount / 100.0)    # 保存之前的状态用于比较    prev_yang = self.kagi_state.is_yang    prev_yin = self.kagi_state.is_yin    prev_ref_price = self.kagi_state.reference_price    # 更新局部极值    if self.kagi_state.is_uptrend:        if current_close > self.kagi_state.local_maximum:            self.kagi_state.local_maximum = current_close            self.kagi_state.reference_price = current_close    if self.kagi_state.is_downtrend:        if current_close < self.kagi_state.local_minimum:            self.kagi_state.local_minimum = current_close            self.kagi_state.reference_price = current_close    # 复杂反转检测：阳线上升趋势 → 阴线下降趋势（卖出信号）    if (self.kagi_state.is_uptrend and self.kagi_state.is_yang and        current_close <= (self.kagi_state.reference_price - reversal_amt) and        current_close < self.kagi_state.local_minimum):        # 更新状态        self.kagi_state.local_maximum = self.kagi_state.reference_price        self.kagi_state.last_reference_price = self.kagi_state.reference_price        self.kagi_state.reference_price = current_close        self.kagi_state.local_minimum = current_close        self.kagi_state.is_downtrend = True        self.kagi_state.is_uptrend = False        self.kagi_state.is_yang = False        self.kagi_state.is_yin = True    # 复杂反转检测：阴线下降趋势 → 阳线上升趋势（买入信号）    elif (self.kagi_state.is_downtrend and self.kagi_state.is_yin and          current_close >= (self.kagi_state.reference_price + reversal_amt) and          current_close > self.kagi_state.local_maximum):        # 更新状态        self.kagi_state.local_minimum = self.kagi_state.reference_price        self.kagi_state.last_reference_price = self.kagi_state.reference_price        self.kagi_state.reference_price = current_close        self.kagi_state.local_maximum = current_close        self.kagi_state.is_downtrend = False        self.kagi_state.is_uptrend = True        self.kagi_state.is_yang = True        self.kagi_state.is_yin = False    # 根据信号模式生成交易信号    signal = None    kagi_line_changed = abs(self.kagi_state.reference_price - prev_ref_price) > 0    if self.signal_mode == SignalMode.EVERY_KAGI_LINE:        # 每条新 Kagi 线都产生信号        if kagi_line_changed:            signal = 'buy' if self.kagi_state.is_uptrend else 'sell'    elif self.signal_mode == SignalMode.ULTRA_AGGRESSIVE:        # 阴阳转换信号        if not prev_yang and self.kagi_state.is_yang:            signal = 'buy'        elif not prev_yin and self.kagi_state.is_yin:            signal = 'sell'    return signal

仓位管理：基于风险的开仓逻辑

当信号触发时，我们不会盲目开仓，而是遵循一套完整的决策流程：

def calculate_position_size(self, entry_price: float, stop_loss: float) -> float:    """    基于风险计算仓位大小    确保每笔交易风险相同的账户百分比，    无论止损距离有多远    """    # 计算风险金额    amount_at_risk = (self.risk_per_trade_percent / 100.0) * self.current_capital    # 计算止损距离    stop_distance = abs(entry_price - stop_loss)    if stop_distance == 0:        return 0.01  # 最小仓位    # 计算仓位大小    position_size = amount_at_risk / stop_distance    return max(0.01, round(position_size, 2))def open_position(self, direction: str, entry_price: float, entry_time: datetime):    """    开设新仓位（不影响现有仓位）    """    # 检查是否达到最大仓位数限制    if len(self.active_positions) >= self.max_concurrent_positions:        return  # 不再开新仓    # 计算止损价格（带缓冲）    if direction == 'long':        base_stop = self.kagi_state.local_minimum        buffer_amount = base_stop * (self.stop_loss_buffer / 100.0)        stop_loss = base_stop - buffer_amount    else:        base_stop = self.kagi_state.local_maximum        buffer_amount = base_stop * (self.stop_loss_buffer / 100.0)        stop_loss = base_stop + buffer_amount    # 计算仓位大小    size = self.calculate_position_size(entry_price, stop_loss)    # 计算止盈价格    stop_distance = abs(entry_price - stop_loss)    if direction == 'long':        take_profit = entry_price + (self.risk_reward_ratio * stop_distance)    else:        take_profit = entry_price - (self.risk_reward_ratio * stop_distance)    # 创建新仓位    position = Position(        position_id=self.next_position_id,        direction=direction,        entry_price=entry_price,        entry_time=entry_time,        size=size,        stop_loss=stop_loss,        take_profit=take_profit    )    self.active_positions.append(position)    self.next_position_id += 1

独立仓位管理

每个仓位都需要独立检查止损止盈：

def check_and_close_positions(self, bar: pd.Series, bar_index: int):    """    检查并关闭触及止损或止盈的仓位    注意：分别检查最高价和最低价，    避免不现实的成交——不能假设在穿过止损之前就能触及止盈    """    current_time = bar.name    high = bar['high']    low = bar['low']    positions_to_close = []    for pos in self.active_positions:        if not pos.is_active:            continue        exit_reason = None        exit_price = None        if pos.direction == 'long':            # 多头仓位：最低价触及止损，最高价触及止盈            if low <= pos.stop_loss:                exit_reason = 'stop_loss'                exit_price = pos.stop_loss            elif high >= pos.take_profit:                exit_reason = 'take_profit'                exit_price = pos.take_profit        else:            # 空头仓位：最高价触及止损，最低价触及止盈            if high >= pos.stop_loss:                exit_reason = 'stop_loss'                exit_price = pos.stop_loss            elif low <= pos.take_profit:                exit_reason = 'take_profit'                exit_price = pos.take_profit        if exit_reason:            positions_to_close.append((pos, exit_price, exit_reason, current_time, bar_index))    # 关闭所有标记的仓位    for pos, exit_price, exit_reason, exit_time, exit_bar in positions_to_close:        self.close_position(pos, exit_price, exit_time, exit_reason, exit_bar)def close_position(self, position: Position, exit_price: float,                   exit_time: datetime, exit_reason: str, exit_bar_index: int):    """关闭指定仓位并记录交易结果"""    # 计算盈亏    if position.direction == 'long':        pnl_points = exit_price - position.entry_price    else:        pnl_points = position.entry_price - exit_price    pnl = pnl_points * position.size    pnl_percent = (pnl / self.current_capital) * 100.0    # 更新资金    self.current_capital += pnl    # 记录交易    trade = Trade(        position_id=position.position_id,        entry_time=position.entry_time,        exit_time=exit_time,        direction=position.direction,        entry_price=position.entry_price,        exit_price=exit_price,        size=position.size,        pnl=pnl,        pnl_percent=pnl_percent,        exit_reason=exit_reason    )    self.trades.append(trade)    # 移除仓位    position.is_active = False    self.active_positions = [p for p in self.active_positions if p.is_active]

完整回测主循环

def run(self, data: pd.DataFrame) -> pd.DataFrame:    """    执行多仓位回测    参数：        data: 包含 open、high、low、close 列的 DataFrame    返回：        交易记录 DataFrame    """    if len(data) == 0:        raise ValueError("数据不能为空")    print(f"🚀 开始回测，共 {len(data)} 根 K 线...")    print(f"   最大同时持仓：{self.max_concurrent_positions} 个")    print(f"   信号模式：{self.signal_mode.value}")    # 初始化 Kagi 状态    self.initialize_kagi_state(data.iloc[0])    self.equity_curve.append((data.index[0], self.current_capital))    for idx in range(1, len(data)):        bar = data.iloc[idx]        current_time = bar.name        current_close = bar['close']        self.bars_since_last_signal += 1        # 检查并关闭触及止损/止盈的仓位        self.check_and_close_positions(bar, idx)        # 获取 Kagi 信号        signal = self.update_kagi_chart(bar)        # 检查是否可以开新仓        can_trade = len(self.active_positions) < self.max_concurrent_positions        # 执行信号        if signal and can_trade:            if signal == 'buy':                self.open_position('long', current_close, current_time)                self.bars_since_last_signal = 0            elif signal == 'sell':                self.open_position('short', current_close, current_time)                self.bars_since_last_signal = 0        # 记录权益        self.equity_curve.append((current_time, self.current_capital))    print(f"✅ 回测完成：共执行 {len(self.trades)} 笔交易")    return self.get_performance_report()

实战案例：使用模拟数据回测

import numpy as np# 生成模拟数据np.random.seed(42)dates = pd.date_range(start='2024-01-01', periods=2000, freq='H')returns = np.random.normal(0.0001, 0.008, len(dates))price = 100 * (1 + returns).cumprod()data = pd.DataFrame({    'open': price * (1 + np.random.uniform(-0.003, 0.003, len(dates))),    'high': price * (1 + np.random.uniform(0, 0.008, len(dates))),    'low': price * (1 + np.random.uniform(-0.008, 0, len(dates))),    'close': price}, index=dates)# 确保 high 是最高价，low 是最低价data['high'] = data[['open', 'high', 'close']].max(axis=1)data['low'] = data[['open', 'low', 'close']].min(axis=1)# 创建回测器backtester = MultiPositionKagiBacktester(    reversal_amount=0.03,           # 0.03% 反转阈值（非常敏感）    max_concurrent_positions=20,    # 最多 20 个同时持仓    risk_per_trade_percent=0.3,     # 每笔交易风险 0.3%    risk_reward_ratio=1.5,          # 风险回报比 1:1.5    signal_mode=SignalMode.EVERY_KAGI_LINE,    stop_loss_buffer=3.0,           # 止损缓冲 3%    initial_capital=10000.0)# 运行回测results = backtester.run(data)# 打印结果print(f"\n💰 初始资金：${backtester.initial_capital:,.2f}")print(f"💰 最终资金：${backtester.current_capital:,.2f}")print(f"📈 收益率：{((backtester.current_capital - backtester.initial_capital) / backtester.initial_capital * 100):.2f}%")print(f"📊 总交易数：{len(backtester.trades)}")

回测结果示例：

🚀 MULTI-POSITION KAGI BACKTEST RESULTS══════════════════════════════════════════💰 CAPITAL   Initial: $10,000.00   Final: $10,906.28   Net Profit: $906.28   Return: 9.06%📊 TRADES   Total: 94 🎯   Winning: 61 (64.9%)   Losing: 33 (35.1%)✅ WINS   Average: $30.77   Largest: $44.83❌ LOSSES   Average: $-29.41   Largest: $-30.02📈 PERFORMANCE   Profit Factor: 1.93

关键经验总结

1. 多仓位的优势

传统"一次一仓"的系统会错失很多机会。通过允许多仓位，我们可以捕捉同一趋势的不同阶段，分散入场点，降低单笔交易的影响。

2. 独立管理至关重要

每个仓位都需要自己的生命周期。1 号仓位可能正在追踪移动止损，而 15 号仓位刚刚入场。

3. 风险管理要分层

仓位层面：独立止损/止盈、最大持仓时间、基于止损距离的仓位大小组合层面：最大同时持仓数、方向均衡、整体回撤保护

4. 先回测再实盘

Python 回测器可以在实盘前发现很多问题，每次迭代都能避免潜在的实盘亏损。

总结

本文介绍的多仓位 Kagi 交易系统代表了一种不同的自动化交易思路。我们不是等待"完美"的入场点然后重仓押注，而是开设多个小仓位，让统计规律发挥作用。这就像经营一家拥有多个收入来源的企业，而不是依赖一份大合同。

核心要点回顾：

• • Kagi 图表通过过滤噪音提供清晰的趋势信号
• • 多仓位同时持有可以捕捉更多机会
• • 独立管理让每笔交易拥有自己的生命周期
• • 风险控制防止灾难性损失
• • Python 回测在投入真金白银前验证策略

无论你是完整实现这个系统，还是借鉴这些原则构建自己的策略，多仓位范式都开启了新的可能性。记住：在多仓位系统中，重要的不是每笔交易都正确，而是管理一个概率组合。有些仓位会失败，有些会成功，但通过适当的风险管理，整体应该随时间呈正向趋势。

参考文章

1. 加入专注于财经数据与量化投研的知识星球【数据科学实战】，获取本文完整研究解析、代码实现细节。

财经数据与量化投研知识社区

2026年全面升级已落地！【数据科学实战】知识星球核心权益如下：

1. 1. 双典系统赋能：获赠《财经数据宝典》与《量化投研宝典》完整文档，凝练多年实战经验，构建系统化知识框架；
2. 2. 量化因子日更教程（2026重磅新增）：每日更新「量化因子专题教程」，配套完整可运行代码与实战案例，深度拆解因子构建、回测与优化全流程；
3. 3. 量化文章专题教程库：300+篇星球独有高质量教程式文章，系统覆盖策略开发、因子研究、风险管理等核心领域，内容基本每日更新，并配套精选学习资料与实战参考；
4. 4. PyBroker实战课程：赠送《PyBroker-入门及实战》视频课程，手把手教学，快速掌握量化策略开发技能；
5. 5. 财经数据支持：定期更新国内外财经数据，为策略研发提供精准、可靠的数据基础；
6. 6. 顶尖学者与行业专家分享：年度邀请学术界博士与业界资深专家开展前沿论文精讲与实战案例分享，不少于4场，直击研究前沿与产业实践；专家直连答疑：与核心开发者及领域专家实时互动，高效解决投研实战难题；
7. 7. 专业社群与专属福利：加入高质量交流社群，获取课程折扣及更多独家资源。

星球已沉淀丰富内容生态——涵盖量化文章专题教程库、因子日更系列、高频数据集、PyBroker实战课程、专家深度分享与实时答疑服务。无论您是初探量化的学习者，还是深耕领域的从业者，这里都是助您少走弯路、高效成长的理想平台。诚邀加入，共探数据驱动的投资未来！

好文推荐