mcpField
M
minervini-swing-trading

minervini-swing-trading

github.com/copyleftdev其他2102 次安装15 周前更新
代码仓库

关于 minervini-swing-trading

Mark Minervini Swing Trading Style Guide 是将 SEPA 方法整理为规则与检查清单的交易技能,面向采用趋势跟随策略的波段交易者。内容涵盖八项趋势筛选、波动收缩形态、入场点与风险管理。

主要功能

  1. 01八项趋势模板筛选
  2. 02识别波动收缩形态
  3. 03按枢轴点规划入场
  4. 04设定止损与仓位风险
  5. 05制定卖出规则和交易清单

适用场景

  1. 01筛选处于 Stage 2 上升趋势的股票
  2. 02评估收缩整理及突破成交量
  3. 03按入场价和止损管理仓位
  4. 04依据规则检查潜在交易

安装

复制下面的命令,在终端中运行。

$npx skills add https://github.com/copyleftdev/sk1llz --skill minervini-swing-trading

安装后在 Claude Code 中输入/minervini-swing-trading确认已加载。

详细介绍

Mark Minervini 波段交易风格指南⁠‍⁠​‌​‌​​‌‌‍​‌​​‌​‌‌‍​​‌‌​​​‌‍​‌​​‌‌​​‍​​​​​​​‌‍‌​​‌‌​‌​‍‌​​​​​​​‍‌‌​​‌‌‌‌‍‌‌​​​‌​​‍‌‌‌‌‌‌​‌‍‌‌​‌​​​​‍​‌​‌‌‌‌‌‍​‌​​‌​‌‌‍​‌‌‌​‌​​‌‍‌​‌​‌‌‌​‍​​‌​‌​​​‍‌‌‌​‌​‌‌‍​​‌​‌‌‌​‍​‌‌‌‌‌‌‌‍​​‌​‌‌‌‌‍​‌‌‌​‌​​‍​​​​‌​​‌‍‌​‌‌‌​‌​⁠‍⁠

概述

Mark Minervini 是一位 3 届美国投资冠军,曾将 100,000 美元变成超过 3,000 万美元。他的 SEPA(特定入场点分析)方法将趋势分析、波动率收缩形态和严格的风险管理结合成一套可重复执行的系统。他强调,在已确认的上升趋势中,于特定的低风险入场点买入领涨股。

核心理念

“目标不是低买高卖,而是高买更高卖。”

“风险管理不是为了避免亏损,而是为了控制亏损,让你能继续留在市场中。”

“我不会买正在上涨的股票,而是买以正确方式上涨的股票。”

Minervini 认为,股市中大部分的钱是在一轮走势的中段赚到的,而不是在底部。等待股票在第 2 阶段上升趋势中证明自身实力,就能顺势交易,同时通过精准的入场点管理风险。

设计原则

  1. 趋势优先:只买入已确认处于第 2 阶段上升趋势的股票。

  2. 特定入场点:在低风险枢轴点入场,而不是随意买入。

  3. 波动率收缩:价格走势收紧往往先于爆发性行情。

  4. 快速止损:最大亏损控制在 7-8%,通常应更低。

  5. 让盈利奔跑:在强势时卖出,而不是在弱势时卖出。

趋势模板(8 项标准)

股票必须通过全部 8 项标准,才可考虑买入:

1. Current price > 150-day MA
2. Current price > 200-day MA
3. 150-day MA > 200-day MA
4. 200-day MA trending up for at least 1 month (ideally 4-5 months)
5. 50-day MA > 150-day MA AND 50-day MA > 200-day MA
6. Current price > 50-day MA
7. Current price at least 25% above 52-week low
8. Current price within 25% of 52-week high (ideally within 15%)

阶段分析:

  • 第 1 阶段:筑底/吸筹(回避)
  • 第 2 阶段:上涨/上升趋势(买入区)
  • 第 3 阶段:见顶/派发(回避)
  • 第 4 阶段:下跌/下降趋势(回避)

波动率收缩形态(VCP)

VCP 是 Minervini 标志性的形态:

VCP Structure:
                    
Price    T1
  |      /\
  |     /  \   T2
  |    /    \ /\   T3
  |   /      X  \ /\    Pivot
  |  /       |   X  \  /----→ BREAKOUT
  | /        |   |   \/
  |/         |   |    
  Base       C1  C2   C3 (contractions tighten)

T = Thrust (price expansion)
C = Contraction (price tightening)

VCP 特征:

  • 至少收缩 2 次,理想情况为 3-4 次
  • 每次收缩幅度都比前一次更小
  • 收缩幅度:第 1 次:20-35%,第 2 次:10-20%,第 3 次:5-15%,第 4 次:3-8%
  • 收缩期间成交量减少(卖盘供应逐渐枯竭)
  • 突破时成交量增加(买盘需求回归)

波段交易时

始终做到
  • 确认股票符合全部 8 项趋势模板标准
  • 等待合适的 VCP 或建设性底部形态
  • 在放量突破枢轴点时入场(成交量比平均水平高 50% 以上)
  • 将止损设在最大亏损 7-8% 的位置(根据形态结构,能更低则更低)
  • 入场前制定卖出计划
  • 交易流动性好的股票(平均成交量 > 400K)
绝不做
  • 买入处于第 1、3 或 4 阶段的股票
  • 追涨过度延伸的股票(高于枢轴点 >10%)
  • 对亏损仓位加仓摊平
  • 持仓承受 8% 以上的亏损
  • 买入缩量突破的股票
  • 忽略相对于大盘的相对强度
优先选择
  • RS 评级 > 85 的股票(表现排名前 15%)
  • 最近几个季度 EPS 增长 > 25% 的股票
  • 紧密盘整(VCP),而非宽幅松散的底部形态
  • 从 IPO 底部形态或首次第 2 阶段突破中走出的股票
  • 行业组强势(行业组排名前 20%)
  • 机构吸筹迹象(放量上涨周)

代码模式

趋势模板扫描器
class TrendTemplateScanner:
    """
    Minervini Trend Template: all 8 criteria must pass.
    This is the first filter—non-negotiable.
    """
    
    def check_trend_template(self, 
                              df: pd.DataFrame,
                              min_200ma_uptrend_days: int = 22) -> TrendTemplateResult:
        """
        Check if stock passes all 8 trend template criteria.
        """
        close = df['close']
        
        # Calculate moving averages
        ma_50 = close.rolling(50).mean()
        ma_150 = close.rolling(150).mean()
        ma_200 = close.rolling(200).mean()
        
        current_price = close.iloc[-1]
        current_50ma = ma_50.iloc[-1]
        current_150ma = ma_150.iloc[-1]
        current_200ma = ma_200.iloc[-1]
        
        # 52-week high/low
        high_52w = close.rolling(252).max().iloc[-1]
        low_52w = close.rolling(252).min().iloc[-1]
        
        # Check 200-day MA trend
        ma_200_month_ago = ma_200.iloc[-min_200ma_uptrend_days]
        ma_200_trending_up = current_200ma > ma_200_month_ago
        
        criteria = {
            '1_price_above_150ma': current_price > current_150ma,
            '2_price_above_200ma': current_price > current_200ma,
            '3_150ma_above_200ma': current_150ma > current_200ma,
            '4_200ma_trending_up': ma_200_trending_up,
            '5_50ma_above_150_and_200': (current_50ma > current_150ma) and (current_50ma > current_200ma),
            '6_price_above_50ma': current_price > current_50ma,
            '7_price_25pct_above_52w_low': current_price >= low_52w * 1.25,
            '8_price_within_25pct_of_52w_high': current_price >= high_52w * 0.75,
        }
        
        all_pass = all(criteria.values())
        
        return TrendTemplateResult(
            passes=all_pass,
            criteria=criteria,
            stage=self.determine_stage(df, criteria),
            price=current_price,
            ma_50=current_50ma,
            ma_150=current_150ma,
            ma_200=current_200ma,
            pct_from_52w_high=(current_price - high_52w) / high_52w * 100,
            pct_from_52w_low=(current_price - low_52w) / low_52w * 100
        )
    
    def determine_stage(self, df: pd.DataFrame, criteria: dict) -> int:
        """
        Determine Weinstein Stage (1-4).
        """
        if all(criteria.values()):
            return 2  # Stage 2 uptrend
        
        close = df['close']
        ma_200 = close.rolling(200).mean()
        
        # Stage 4: Price below declining 200 MA
        if close.iloc[-1] < ma_200.iloc[-1] and ma_200.iloc[-1] < ma_200.iloc[-22]:
            return 4
        
        # Stage 3: Price near/below flattening 200 MA
        if close.iloc[-1] < ma_200.iloc[-1] * 1.05:
            return 3
        
        # Stage 1: Basing
        return 1
    
    def scan_universe(self, 
                      symbols: List[str], 
                      data: Dict[str, pd.DataFrame]) -> List[TrendTemplateResult]:
        """
        Scan universe for stocks passing trend template.
        """
        results = []
        
        for symbol in symbols:
            df = data[symbol]
            if len(df) < 200:  # Need enough history
                continue
            
            result = self.check_trend_template(df)
            result.symbol = symbol
            
            if result.passes:
                results.append(result)
        
        # Sort by proximity to 52-week high (tighter = better)
        return sorted(results, key=lambda x: x.pct_from_52w_high, reverse=True)
VCP 形态检测器
class VCPDetector:
    """
    Volatility Contraction Pattern detection.
    The tighter the contractions, the more explosive the breakout.
    """
    
    def __init__(self, 
                 min_contractions: int = 2,
                 max_first_contraction: float = 0.35,
                 contraction_ratio: float = 0.6):
        self.min_contractions = min_contractions
        self.max_first_contraction = max_first_contraction
        self.contraction_ratio = contraction_ratio  # Each contraction should be this % of previous
    
    def detect_vcp(self, df: pd.DataFrame) -> VCPResult:
        """
        Detect VCP pattern in price data.
        """
        close = df['close']
        high = df['high']
        low = df['low']
        volume = df['volume']
        
        # Find recent high (potential left side of base)
        lookback = 60  # ~3 months
        recent_high_idx = high.iloc[-lookback:].idxmax()
        recent_high = high.loc[recent_high_idx]
        
        # Find contractions from that high
        contractions = self.find_contractions(df, recent_high_idx)
        
        if len(contractions) < self.min_contractions:
            return VCPResult(valid=False, reason="Insufficient contractions")
        
        # Validate contraction depths are decreasing
        if not self.validate_contraction_depths(contractions):
            return VCPResult(valid=False, reason="Contractions not tightening")
        
        # Check volume pattern (should decrease during base)
        if not self.validate_volume_pattern(df, recent_high_idx):
            return VCPResult(valid=False, reason="Volume not contracting")
        
        # Calculate pivot point
        pivot = self.calculate_pivot(df, contractions)
        
        # Calculate tightness score (lower is better)
        tightness = contractions[-1]['depth']
        
        return VCPResult(
            valid=True,
            contractions=contractions,
            pivot_price=pivot,
            tightness_pct=tightness * 100,
            base_length_days=(df.index[-1] - df.index[recent_high_idx]).days,
            volume_dry_up=self.calculate_volume_dryup(df, recent_high_idx)
        )
    
    def find_contractions(self, 
                          df: pd.DataFrame, 
                          start_idx) -> List[dict]:
        """
        Find swing high/low contractions from start point.
        """
        high = df['high']
        low = df['low']
        
        contractions = []
        current_high = high.loc[start_idx]
        
        # Walk forward finding contractions
        subset = df.loc[start_idx:]
        
        i = 0
        while i < len(subset) - 5:
            # Find next swing low
            window = subset.iloc[i:i+10]
            swing_low_idx = window['low'].idxmin()
            swing_low = window['low'].loc[swing_low_idx]
            
            # Find next swing high after that
            remaining = subset.loc[swing_low_idx:]
            if len(remaining) < 5:
                break
            
            next_window = remaining.iloc[:10]
            swing_high_idx = next_window['high'].idxmax()
            swing_high = next_window['high'].loc[swing_high_idx]
            
            depth = (current_high - swing_low) / current_high
            
            contractions.append({
                'high': current_high,
                'low': swing_low,
                'depth': depth,
                'high_date': start_idx if len(contractions) == 0 else swing_high_idx,
                'low_date': swing_low_idx
            })
            
            current_high = swing_high
            i = subset.index.get_loc(swing_high_idx) - subset.index.get_loc(subset.index[0])
            i += 1
        
        return contractions
    
    def validate_contraction_depths(self, contractions: List[dict]) -> bool:
        """
        Each contraction should be shallower than the previous.
        """
        for i in range(1, len(contractions)):
            if contractions[i]['depth'] >= contractions[i-1]['depth'] * 1.1:  # Allow 10% tolerance
                return False
        return True
    
    def validate_volume_pattern(self, df: pd.DataFrame, start_idx) -> bool:
        """
        Volume should decrease during the base formation.
        """
        volume = df['volume']
        subset = volume.loc[start_idx:]
        
        if len(subset) < 20:
            return False
        
        first_half_avg = subset.iloc[:len(subset)//2].mean()
        second_half_avg = subset.iloc[len(subset)//2:].mean()
        
        return second_half_avg < first_half_avg * 0.9  # Volume should be lower
    
    def calculate_pivot(self, df: pd.DataFrame, contractions: List[dict]) -> float:
        """
        Pivot is the high of the last contraction.
        """
        if not contractions:
            return df['high'].iloc[-20:].max()
        
        return contractions[-1]['high']
    
    def calculate_volume_dryup(self, df: pd.DataFrame, start_idx) -> float:
        """
        How much has volume dried up during the base?
        """
        volume = df['volume']
        avg_volume_before = volume.loc[:start_idx].iloc[-20:].mean()
        recent_volume = volume.iloc[-5:].mean()
        
        return (avg_volume_before - recent_volume) / avg_volume_before
入场与风险管理
class MinerviniTradeManager:
    """
    Entry, position sizing, and risk management per Minervini rules.
    """
    
    def __init__(self, 
                 account_size: float,
                 max_risk_per_trade: float = 0.01,  # 1%
                 max_position_pct: float = 0.25):    # 25% max single position
        self.account = account_size
        self.risk_per_trade = max_risk_per_trade
        self.max_position = max_position_pct
    
    def calculate_entry(self, 
                        vcp: VCPResult, 
                        current_price: float) -> EntryPlan:
        """
        Calculate entry point and buy zone.
        """
        pivot = vcp.pivot_price
        
        # Buy zone: pivot to 5% above pivot
        buy_zone_low = pivot
        buy_zone_high = pivot * 1.05
        
        # Is current price in buy zone?
        in_buy_zone = buy_zone_low <= current_price <= buy_zone_high
        
        # Extended if >5% above pivot
        extended = current_price > buy_zone_high
        
        return EntryPlan(
            pivot_price=pivot,
            buy_zone=(buy_zone_low, buy_zone_high),
            current_price=current_price,
            in_buy_zone=in_buy_zone,
            extended=extended,
            pct_above_pivot=(current_price - pivot) / pivot * 100
        )
    
    def calculate_stop(self, 
                       entry_price: float, 
                       vcp: VCPResult,
                       max_stop_pct: float = 0.08) -> StopPlan:
        """
        Calculate stop loss based on chart structure.
        Minervini: max 7-8%, but tighter if structure allows.
        """
        # Option 1: Below the last contraction low
        structure_stop = vcp.contractions[-1]['low'] * 0.99  # 1% below
        structure_stop_pct = (entry_price - structure_stop) / entry_price
        
        # Option 2: Fixed percentage
        fixed_stop = entry_price * (1 - max_stop_pct)
        
        # Use tighter of the two
        if structure_stop_pct <= max_stop_pct:
            stop_price = structure_stop
            stop_type = 'STRUCTURE'
        else:
            stop_price = fixed_stop
            stop_type = 'FIXED_PCT'
        
        return StopPlan(
            stop_price=stop_price,
            stop_pct=(entry_price - stop_price) / entry_price * 100,
            stop_type=stop_type,
            structure_stop=structure_stop,
            fixed_stop=fixed_stop
        )
    
    def calculate_position_size(self, 
                                 entry_price: float, 
                                 stop_price: float) -> PositionSize:
        """
        Position sizing based on risk.
        """
        risk_amount = self.account * self.risk_per_trade
        risk_per_share = entry_price - stop_price
        
        # Shares based on risk
        shares_by_risk = int(risk_amount / risk_per_share)
        
        # Max position check
        max_shares = int(self.account * self.max_position / entry_price)
        
        final_shares = min(shares_by_risk, max_shares)
        
        return PositionSize(
            shares=final_shares,
            position_value=final_shares * entry_price,
            position_pct=final_shares * entry_price / self.account * 100,
            risk_dollars=final_shares * risk_per_share,
            risk_pct=final_shares * risk_per_share / self.account * 100,
            limited_by='RISK' if shares_by_risk < max_shares else 'MAX_POSITION'
        )
    
    def create_trade_plan(self, 
                          symbol: str,
                          df: pd.DataFrame,
                          vcp: VCPResult) -> TradePlan:
        """
        Complete trade plan with entry, stop, and targets.
        """
        current_price = df['close'].iloc[-1]
        
        entry = self.calculate_entry(vcp, current_price)
        stop = self.calculate_stop(entry.pivot_price, vcp)
        position = self.calculate_position_size(entry.pivot_price, stop.stop_price)
        
        # Profit targets
        risk = entry.pivot_price - stop.stop_price
        target_1 = entry.pivot_price + (risk * 2)   # 2:1
        target_2 = entry.pivot_price + (risk * 3)   # 3:1
        target_3 = entry.pivot_price * 1.20         # 20% move
        
        return TradePlan(
            symbol=symbol,
            entry=entry,
            stop=stop,
            position=position,
            targets={
                '2R': target_1,
                '3R': target_2,
                '20%': target_3
            },
            risk_reward_ratio=2.0,  # Minimum acceptable
            breakout_volume_required=df['volume'].rolling(50).mean().iloc[-1] * 1.5
        )
卖出规则
class MinerviniSellRules:
    """
    Minervini's selling discipline: protect gains, cut losses.
    """
    
    def check_sell_signals(self, 
                           trade: ActiveTrade,
                           df: pd.DataFrame) -> List[SellSignal]:
        """
        Check all sell rules and return triggered signals.
        """
        signals = []
        current_price = df['close'].iloc[-1]
        
        # 1. STOP LOSS (mandatory)
        if current_price <= trade.stop_price:
            signals.append(SellSignal(
                type='STOP_LOSS',
                priority=1,
                action='SELL_ALL',
                reason=f'Price {current_price:.2f} hit stop {trade.stop_price:.2f}'
            ))
        
        # 2. Climax top (sell into strength)
        if self.detect_climax_top(df, trade):
            signals.append(SellSignal(
                type='CLIMAX_TOP',
                priority=2,
                action='SELL_HALF',
                reason='Climactic price/volume action'
            ))
        
        # 3. Break of 50-day MA after extended run
        if self.check_50ma_break(df, trade):
            signals.append(SellSignal(
                type='50MA_BREAK',
                priority=3,
                action='SELL_HALF',
                reason='Closed below 50-day MA after extended move'
            ))
        
        # 4. Lower low after lower high (trend change)
        if self.detect_lower_low(df):
            signals.append(SellSignal(
                type='TREND_CHANGE',
                priority=2,
                action='SELL_ALL',
                reason='Lower high followed by lower low'
            ))
        
        # 5. Holding period too long without progress
        if self.check_stalled_trade(trade, current_price):
            signals.append(SellSignal(
                type='TIME_STOP',
                priority=4,
                action='REVIEW',
                reason='Position stalled for 3+ weeks'
            ))
        
        return sorted(signals, key=lambda x: x.priority)
    
    def detect_climax_top(self, df: pd.DataFrame, trade: ActiveTrade) -> bool:
        """
        Climax top: largest single-day gain on highest volume.
        Often signals exhaustion.
        """
        close = df['close']
        volume = df['volume']
        
        # Recent daily returns
        daily_return = close.pct_change().iloc[-1]
        avg_return = close.pct_change().iloc[-50:].mean()
        
        # Volume comparison
        current_volume = volume.iloc[-1]
        avg_volume = volume.rolling(50).mean().iloc[-1]
        
        # Climax: big up day (>2x average return) on huge volume (>2x average)
        is_climax = (daily_return > avg_return * 3) and (current_volume > avg_volume * 2)
        
        # Only matters if we're already up significantly
        current_gain = (close.iloc[-1] - trade.entry_price) / trade.entry_price
        
        return is_climax and current_gain > 0.20
    
    def check_50ma_break(self, df: pd.DataFrame, trade: ActiveTrade) -> bool:
        """
        Close below 50 MA after being extended above it.
        """
        close = df['close']
        ma_50 = close.rolling(50).mean()
        
        current_price = close.iloc[-1]
        current_50ma = ma_50.iloc[-1]
        
        # Was extended above 50 MA?
        max_extension = ((close.iloc[-20:] - ma_50.iloc[-20:]) / ma_50.iloc[-20:]).max()
        
        return current_price < current_50ma and max_extension > 0.10
    
    def detect_lower_low(self, df: pd.DataFrame) -> bool:
        """
        Lower high followed by lower low = potential trend change.
        """
        high = df['high']
        low = df['low']
        
        # Find recent swing points
        recent_high_1 = high.iloc[-20:-10].max()
        recent_high_2 = high.iloc[-10:].max()
        recent_low_1 = low.iloc[-20:-10].min()
        recent_low_2 = low.iloc[-10:].min()
        
        lower_high = recent_high_2 < recent_high_1
        lower_low = recent_low_2 < recent_low_1
        
        return lower_high and lower_low
    
    def check_stalled_trade(self, 
                            trade: ActiveTrade, 
                            current_price: float,
                            max_stall_days: int = 15) -> bool:
        """
        Position going nowhere for too long.
        """
        days_held = (datetime.now() - trade.entry_date).days
        gain_pct = (current_price - trade.entry_price) / trade.entry_price
        
        # Stalled: held >15 days with <5% gain
        return days_held > max_stall_days and gain_pct < 0.05

思维模型

Minervini 通过以下问题来进行波段交易:

  1. 是否处于第 2 阶段? 如果不是,就完全跳过
  2. 是否有合适的底部形态? VCP 或建设性形态
  3. 枢轴点在哪里? 明确入场点并界定风险
  4. 我的风险是什么? 始终在入场前设定止损
  5. 我入场过早还是过晚? 只在买入区买入,绝不追高

交易检查清单

□ Stock passes ALL 8 trend template criteria
□ VCP or proper base pattern identified
□ Volume declining during base (supply dried up)
□ Pivot point clearly defined
□ Entry within 5% of pivot (not extended)
□ Stop loss set (max 7-8%, tighter if possible)
□ Position sized to 1% account risk
□ Volume surge on breakout (50%+ above average)
□ RS Rating > 80 (top performers)
□ EPS growth positive and accelerating

Minervini 的标志性操作

  • 趋势模板(8 项标准筛选)
  • 波动率收缩形态(VCP)
  • 只做第 2 阶段(绝不做第 1、3 或 4 阶段)
  • 在枢轴点买入,不提前买入
  • 最大止损设为 7-8%
  • 趁强势卖出(高潮顶部)
  • 按风险确定仓位规模
  • 突破时确认成交量

评论

支持 Markdown