【Python 量化实战】手把手教你实现神奇九转策略（附完整可运行代码）

weixin-npy8264926

733人浏览 · 2026-04-27 15:37:58

weixin-npy8264926 · 2026-04-27 15:37:58 发布

一、什么是神奇九转策略？

神奇九转（Nine Rotation Indicator）是目前 A 股市场最受欢迎的技术分析指标之一，其核心理念源于美国技术分析大师 ** 汤姆・迪马克（Tom DeMark）** 于 20 世纪 70 年代提出的 TD 序列理论。

迪马克通过研究超过 10 万小时的全球交易数据发现：当价格连续突破特定阈值时，市场情绪将走向极端，原有趋势动能会快速衰竭，形成高概率的反转临界点。该理论最初应用于美股市场，2000 年后经国内量化团队简化优化，形成了更适配 A 股震荡市特征的 "九转序列" 指标。

根据 2023 年国内头部券商回测数据显示，该策略在沪深 300 成分股中的有效触发率达68.3%，尤其适合震荡市中的波段交易，是散户和量化新手入门的首选策略之一。

二、神奇九转核心理论与规则

2.1 三大核心定律

神奇九转的底层逻辑基于价格波动的三大客观规律：

趋势衰竭定律：连续 9 日价格突破前 4 日基准价，表明原有趋势动能已基本耗尽
情绪极端定律：当市场参与者一致性预期连续 9 次得到验证时，反身性效应将引发趋势反转
周期嵌套定律：日线级别的九转信号往往对应 120 分钟周期的反向序列完成，多周期共振可大幅提高胜率

2.2 双重验证机制

为过滤假信号，标准神奇九转加入了严格的双重验证规则：

上涨九转（卖出信号）：第 8、9 日的最高价必须突破第 6、7 日的最高价峰值
下跌九转（买入信号）：第 8、9 日的最低价必须跌破第 6、7 日的最低价谷值

2.3 基础计数规则

连续 9 根 K 线收盘价高于前第 4 日收盘价 → 触发上涨九转（潜在卖出信号）
连续 9 根 K 线收盘价低于前第 4 日收盘价 → 触发下跌九转（潜在买入信号）
计数从第 6 个连续信号开始显性标记，一旦连续条件中断，计数立即重置为 0

三、完整 Python 代码实现

3.1 环境依赖

首先安装所需的第三方库：

pip install akshare pandas streamlit plotly

3.2 核心算法实现

# -*- coding: utf-8 -*-
import akshare as ak
import pandas as pd
import streamlit as st
import plotly.graph_objects as go
from datetime import datetime

def calculate_nine_turns(df):
    """
    计算神奇九转序列
    参数：
        df: 包含open, high, low, close列的DataFrame
    返回：
        新增九转序列和买卖信号列的DataFrame
    """
    # 计算涨跌条件
    df['up_condition'] = df['close'] > df['close'].shift(4)
    df['down_condition'] = df['close'] < df['close'].shift(4)
    
    # 连续计数逻辑
    for condition in ['up', 'down']:
        streak = 0
        streaks = []
        for idx in range(len(df)):
            if df[f"{condition}_condition"].iloc[idx]:
                streak = streak + 1 if streak < 9 else 9
            else:
                streak = 0
            streaks.append(streak)
        df[f'{condition}_streak'] = streaks
    
    # 生成基础买卖信号（可结合均线进一步过滤）
    df['ma20'] = df['close'].rolling(20).mean()
    # 基础版：仅用九转信号
    df['buy_signal'] = (df['down_streak'] == 9)
    df['sell_signal'] = (df['up_streak'] == 9)
    
    # 进阶版：结合20日均线过滤假信号（推荐）
    # df['buy_signal'] = (df['down_streak'] == 9) & (df['close'] > df['ma20'])
    # df['sell_signal'] = (df['up_streak'] == 9) & (df['close'] < df['ma20'])
    
    return df

3.3 A 股历史数据获取

使用 akshare 获取 A 股前复权日线数据：

def get_stock_data(symbol, start_date, end_date):
    """
    获取A股前复权日线数据
    参数：
        symbol: 股票代码（如600120）
        start_date: 开始日期（datetime对象）
        end_date: 结束日期（datetime对象）
    返回：
        标准化的行情DataFrame
    """
    try:
        df = ak.stock_zh_a_hist(
            symbol=symbol,
            period="daily",
            start_date=start_date.strftime("%Y%m%d"),
            end_date=end_date.strftime("%Y%m%d"),
            adjust="qfq"
        )
        
        # 列名标准化
        df = df.rename(columns={
            '日期': 'date',
            '开盘': 'open',
            '收盘': 'close',
            '最高': 'high',
            '最低': 'low',
            '成交量': 'volume'
        })
        
        df['date'] = pd.to_datetime(df['date'])
        return df.set_index('date').sort_index()
    
    except Exception as e:
        st.error(f"数据获取失败: {str(e)}")
        return None

3.4 K 线图 + 信号可视化

使用 plotly 绘制交互式 K 线图，并标注九转序列和买卖信号：

def plot_kline_with_signals(df):
    """
    绘制带神奇九转标记和买卖信号的K线图
    """
    fig = go.Figure()
    
    # 绘制主K线
    fig.add_trace(go.Candlestick(
        x=df.index,
        open=df['open'],
        high=df['high'],
        low=df['low'],
        close=df['close'],
        increasing_line_color='#FF4500',
        decreasing_line_color='#1E90FF',
        name='K线'
    ))
    
    # 标记九转序列（从第6个开始显示）
    up_points = df[df['up_streak'] >= 6]
    down_points = df[df['down_streak'] >= 6]
    
    # 上涨九转标记（红色）
    fig.add_trace(go.Scatter(
        x=up_points.index,
        y=up_points['high'] * 1.02,
        mode='text',
        text=up_points['up_streak'].astype(str),
        textfont=dict(color='#FF4500', size=14),
        name='上涨九转'
    ))
    
    # 下跌九转标记（蓝色）
    fig.add_trace(go.Scatter(
        x=down_points.index,
        y=down_points['low'] * 0.98,
        mode='text',
        text=down_points['down_streak'].astype(str),
        textfont=dict(color='#1E90FF', size=14),
        name='下跌九转'
    ))
    
    # 标记买卖信号
    buy_signals = df[df['buy_signal']]
    sell_signals = df[df['sell_signal']]
    
    # 买入信号（绿色向上三角）
    fig.add_trace(go.Scatter(
        x=buy_signals.index,
        y=buy_signals['low'] * 0.95,
        mode='markers',
        marker=dict(color='green', size=12, symbol='triangle-up'),
        name='买入信号'
    ))
    
    # 卖出信号（红色向下三角）
    fig.add_trace(go.Scatter(
        x=sell_signals.index,
        y=sell_signals['high'] * 1.05,
        mode='markers',
        marker=dict(color='red', size=12, symbol='triangle-down'),
        name='卖出信号'
    ))
    
    # 图表布局设置
    fig.update_layout(
        title='神奇九转策略K线图',
        xaxis_title='日期',
        yaxis_title='价格',
        height=600,
        xaxis_rangeslider_visible=False,
        hovermode='x unified'
    )
    
    return fig

3.5 主程序入口（Streamlit 界面）

def main():
    st.set_page_config(page_title='神奇九转策略工具', layout='wide')
    st.title('神奇九转策略可视化工具')
    
    # 侧边栏参数设置
    with st.sidebar:
        st.header('参数设置')
        symbol = st.text_input('股票代码', value='600120')
        start_date = st.date_input('开始日期', value=datetime(2025, 1, 1))
        end_date = st.date_input('结束日期', value=datetime.now())
        
        if st.button('运行策略'):
            st.session_state.run = True
    
    # 运行策略并展示结果
    if 'run' in st.session_state and st.session_state.run:
        with st.spinner('正在获取数据并计算...'):
            df = get_stock_data(symbol, start_date, end_date)
            
            if df is not None and not df.empty:
                df = calculate_nine_turns(df)
                fig = plot_kline_with_signals(df)
                
                st.plotly_chart(fig, use_container_width=True)
                
                # 展示信号统计
                buy_count = df['buy_signal'].sum()
                sell_count = df['sell_signal'].sum()
                st.subheader('信号统计')
                col1, col2 = st.columns(2)
                col1.metric('买入信号次数', buy_count)
                col2.metric('卖出信号次数', sell_count)
                
                # 展示原始数据
                with st.expander('查看完整数据'):
                    st.dataframe(df)

if __name__ == '__main__':
    main()