☰
Python实战:技术分析框架与量化交易信号识别全流程
2026/9/26 20:26:52 网站建设 项目流程

最近在跟踪科技板块走势时,很多朋友都注意到了“东山精密”、“通富微电”、“深南电路”这几只标的在特定时点的异动,并好奇背后的分析逻辑。同时,“汇金”的动向也牵动着市场神经。本文将从一个技术分析爱好者的视角,系统性拆解这类问题背后的分析框架、常用工具和数据验证方法,重点不在于预测明天涨跌,而在于分享一套可复现、可验证的技术分析实战流程。无论你是刚接触股票数据分析的程序员,还是希望用量化思维辅助决策的投资者,都能从中获得从数据获取、处理、指标计算到可视化分析的全流程代码实践。

1. 技术分析的核心逻辑与基础概念

在讨论具体标的之前,必须明确我们讨论的“分析”建立在何种假设之上。这里的“逻辑”通常指基于市场公开数据(价格、成交量)的技术分析逻辑,而非公司基本面。

技术分析的核心假设:

  1. 市场行为涵盖一切信息:所有已知的、未知的基本面、政策面、心理面因素都已反映在价格和成交量中。
  2. 价格沿趋势运动:趋势一旦形成,更倾向于延续而非反转。
  3. 历史会重演:市场参与者的心理和行为模式会在相似的市场条件下重复出现。

基于以上假设,分析“为什么可以反弹”通常是在寻找趋势转换或支撑确认的信号。对于“东山”、“通富”、“深南”这类科技股,其波动性往往高于大盘,技术信号的灵敏度也更高。

关键分析维度:

  • 趋势分析:判断当前处于上升、下降还是盘整趋势。
  • 支撑与阻力:识别价格可能反弹(支撑)或回落(阻力)的关键价位。
  • 动量指标:衡量价格上涨或下跌的速度和力量,判断趋势的强弱。
  • 成交量验证:价格变动是否得到成交量的支持,是判断信号有效性的重要依据。

接下来,我们将通过Python实战,搭建一个完整的数据分析环境,并逐一实现这些分析维度。

2. 环境准备与数据分析栈搭建

我们使用Python作为分析工具,因为它拥有丰富且成熟的金融数据分析库。

环境与版本说明:

  • 操作系统:Windows 10/11, macOS 或 Linux 均可。
  • Python版本:3.8 及以上。
  • 核心库:
    • pandas(>=1.3.0): 数据处理与分析。
    • numpy(>=1.21.0): 数值计算。
    • matplotlib(>=3.5.0) &mplfinance(>=0.12.7): 金融数据可视化。
    • akshare(>=1.8.0): 免费、开源的财经数据接口库,用于获取A股数据。
    • ta(>=0.10.0): 技术指标库,包含大量常见指标的计算。

项目初始化与依赖安装:

  1. 创建项目目录并初始化虚拟环境(推荐)。
    mkdir stock_tech_analysis && cd stock_tech_analysis python -m venv venv # Windows激活: venv\Scripts\activate # macOS/Linux激活: source venv/bin/activate
  2. 安装所需依赖库。
    pip install pandas numpy matplotlib mplfinance akshare ta -i https://pypi.tuna.tsinghua.edu.cn/simple

示例项目结构:

stock_tech_analysis/ ├── data/ # 缓存数据目录 ├── utils/ # 工具函数 │ └── data_fetcher.py ├── indicators/ # 自定义指标计算 │ └── trend_analyzer.py ├── config.py # 配置文件(如股票代码列表) └── main_analysis.py # 主分析脚本

3. 数据获取与预处理实战

一切分析始于数据。我们使用akshare获取可靠的历史行情数据。

3.1 获取单只股票历史数据

首先,我们编写一个数据获取工具函数。

# utils/data_fetcher.py import akshare as ak import pandas as pd import os from datetime import datetime, timedelta def fetch_stock_daily(stock_code, start_date='20230101', end_date=None, adjust='qfq'): """ 获取股票日线行情数据 :param stock_code: 股票代码,如 '002384'(东山精密), '002156'(通富微电) :param start_date: 开始日期,格式 'YYYYMMDD' :param end_date: 结束日期,默认今天 :param adjust: 复权类型,'qfq'(前复权),'hfq'(后复权),''(不复权) :return: pandas DataFrame """ if end_date is None: end_date = datetime.now().strftime('%Y%m%d') # 尝试从缓存读取 cache_file = f'data/{stock_code}_{start_date}_{end_date}_{adjust}.pkl' if os.path.exists(cache_file): print(f'从缓存加载数据: {stock_file}') return pd.read_pickle(cache_file) try: # 使用 ak.stock_zh_a_hist 接口 df = ak.stock_zh_a_hist(symbol=stock_code, period="daily", start_date=start_date, end_date=end_date, adjust=adjust) # 规范列名 df.columns = ['date', 'open', 'close', 'high', 'low', 'volume', 'amount', 'amplitude', 'pct_change', 'change', 'turnover'] df['date'] = pd.to_datetime(df['date']) df.set_index('date', inplace=True) df.sort_index(inplace=True) # 缓存数据 os.makedirs('data', exist_ok=True) df.to_pickle(cache_file) print(f'数据已获取并缓存: {cache_file}') return df except Exception as e: print(f'获取数据失败: {stock_code}, 错误: {e}') return None # 测试获取东山精密数据 if __name__ == '__main__': df_ds = fetch_stock_daily('002384', start_date='20240101') print(df_ds.head()) print(f'数据形状: {df_ds.shape}')

3.2 数据清洗与特征工程

获取的原始数据需要清洗,并计算一些基础特征。

# utils/data_processor.py import pandas as pd import numpy as np def calculate_basic_features(df): """ 计算基础技术特征 """ if df is None or df.empty: return df df = df.copy() # 1. 计算涨跌幅(日级) df['returns'] = df['close'].pct_change() # 2. 计算简单移动平均线 (SMA) df['sma_5'] = df['close'].rolling(window=5).mean() df['sma_10'] = df['close'].rolling(window=10).mean() df['sma_20'] = df['close'].rolling(window=20).mean() # 月线 df['sma_60'] = df['close'].rolling(window=60).mean() # 季线 # 3. 计算指数移动平均线 (EMA) - 对近期价格更敏感 df['ema_12'] = df['close'].ewm(span=12, adjust=False).mean() df['ema_26'] = df['close'].ewm(span=26, adjust=False).mean() # 4. 计算布林带 (Bollinger Bands) df['bb_middle'] = df['close'].rolling(window=20).mean() bb_std = df['close'].rolling(window=20).std() df['bb_upper'] = df['bb_middle'] + 2 * bb_std df['bb_lower'] = df['bb_middle'] - 2 * bb_std df['bb_width'] = (df['bb_upper'] - df['bb_lower']) / df['bb_middle'] # 布林带宽度,衡量波动率 # 5. 计算价格相对位置(用于判断超买超卖) df['position_in_bb'] = (df['close'] - df['bb_lower']) / (df['bb_upper'] - df['bb_lower']) # 6. 计算ATR(平均真实波幅),衡量波动性 high_low = df['high'] - df['low'] high_close_prev = abs(df['high'] - df['close'].shift()) low_close_prev = abs(df['low'] - df['close'].shift()) true_range = pd.concat([high_low, high_close_prev, low_close_prev], axis=1).max(axis=1) df['atr_14'] = true_range.rolling(window=14).mean() return df

4. 核心指标计算与信号识别

有了基础数据,我们可以计算更复杂的技术指标来识别潜在信号。

4.1 动量指标:MACD与RSI

我们使用ta库来高效计算标准指标。

# indicators/momentum_indicators.py import pandas as pd import ta def calculate_macd(df, fast=12, slow=26, signal=9): """计算MACD指标""" df = df.copy() macd_indicator = ta.trend.MACD(close=df['close'], window_slow=slow, window_fast=fast, window_sign=signal) df['macd'] = macd_indicator.macd() df['macd_signal'] = macd_indicator.macd_signal() df['macd_diff'] = macd_indicator.macd_diff() # 柱状图 return df def calculate_rsi(df, window=14): """计算RSI指标""" df = df.copy() df['rsi'] = ta.momentum.RSIIndicator(close=df['close'], window=window).rsi() return df def identify_macd_signals(df): """识别MACD金叉/死叉信号""" df = df.copy() # 金叉:MACD线上穿信号线,且为正值区域更佳 df['macd_golden_cross'] = (df['macd'] > df['macd_signal']) & \ (df['macd'].shift() <= df['macd_signal'].shift()) # 死叉:MACD线下穿信号线 df['macd_death_cross'] = (df['macd'] < df['macd_signal']) & \ (df['macd'].shift() >= df['macd_signal'].shift()) return df def identify_rsi_signals(df, oversold=30, overbought=70): """识别RSI超买超卖信号""" df = df.copy() df['rsi_oversold'] = df['rsi'] < oversold df['rsi_overbought'] = df['rsi'] > overbought # RSI从超卖区上穿30线 df['rsi_bullish_exit'] = (df['rsi'] > oversold) & (df['rsi'].shift() <= oversold) # RSI从超买区下穿70线 df['rsi_bearish_exit'] = (df['rsi'] < overbought) & (df['rsi'].shift() >= overbought) return df

4.2 成交量指标:OBV与量价关系

成交量是验证价格变动的重要工具。

# indicators/volume_indicators.py import pandas as pd import ta def calculate_obv(df): """计算能量潮(OBV)指标""" df = df.copy() df['obv'] = ta.volume.OnBalanceVolumeIndicator(close=df['close'], volume=df['volume']).on_balance_volume() # 计算OBV的EMA,作为信号线 df['obv_ema_20'] = df['obv'].ewm(span=20, adjust=False).mean() # OBV突破其均线,可能预示价格突破 df['obv_break'] = (df['obv'] > df['obv_ema_20']) & (df['obv'].shift() <= df['obv_ema_20'].shift()) return df def analyze_volume_price(df): """分析量价关系""" df = df.copy() # 放量上涨(价增量增) df['volume_up_with_price'] = (df['volume'] > df['volume'].rolling(20).mean() * 1.5) & \ (df['returns'] > 0) # 放量下跌(价跌量增) df['volume_down_with_price'] = (df['volume'] > df['volume'].rolling(20).mean() * 1.5) & \ (df['returns'] < 0) # 缩量调整(价跌量缩) df['volume_shrink_on_down'] = (df['volume'] < df['volume'].rolling(20).mean() * 0.7) & \ (df['returns'] < 0) return df

4.3 趋势与支撑阻力识别

# indicators/trend_support_resistance.py import pandas as pd import numpy as np def identify_support_resistance(df, window=20): """ 简易支撑阻力位识别(基于局部高点和低点) """ df = df.copy() # 识别局部高点(前后window天内最高) df['local_high'] = df['high'].rolling(window=window*2+1, center=True).max() df['is_resistance'] = (df['high'] == df['local_high']) & \ (df['high'] > df['high'].shift(1)) & \ (df['high'] > df['high'].shift(-1)) # 识别局部低点(前后window天内最低) df['local_low'] = df['low'].rolling(window=window*2+1, center=True).min() df['is_support'] = (df['low'] == df['local_low']) & \ (df['low'] < df['low'].shift(1)) & \ (df['low'] < df['low'].shift(-1)) # 提取关键的支撑阻力位(去重,取最近N个) resistance_levels = df[df['is_resistance']]['high'].dropna().unique()[-5:] support_levels = df[df['is_support']]['low'].dropna().unique()[-5:] return df, support_levels, resistance_levels def check_trend_strength(df): """ 评估趋势强度 """ df = df.copy() # 使用ADX(平均趋向指数)衡量趋势强度,这里用简单方法替代 # 趋势向上:价格在短期均线之上,且均线多头排列 df['trend_up'] = (df['close'] > df['sma_20']) & \ (df['sma_5'] > df['sma_10']) & (df['sma_10'] > df['sma_20']) # 趋势向下:价格在短期均线之下,且均线空头排列 df['trend_down'] = (df['close'] < df['sma_20']) & \ (df['sma_5'] < df['sma_10']) & (df['sma_10'] < df['sma_20']) # 趋势强度:近期涨幅与波动率比值 df['trend_strength'] = df['returns'].rolling(5).mean() / df['returns'].rolling(20).std() return df

5. 综合分析与可视化实战

我们将所有分析整合,并对“东山精密”进行一段历史数据的回溯分析,模拟“寻找反弹信号”的过程。

# main_analysis.py import pandas as pd import matplotlib.pyplot as plt import mplfinance as mpf from utils.data_fetcher import fetch_stock_daily from utils.data_processor import calculate_basic_features from indicators.momentum_indicators import calculate_macd, calculate_rsi, identify_macd_signals, identify_rsi_signals from indicators.volume_indicators import calculate_obv, analyze_volume_price from indicators.trend_support_resistance import identify_support_resistance, check_trend_strength def comprehensive_analysis(stock_code='002384', start_date='20240101'): """ 执行综合分析流程 """ print(f'开始分析股票: {stock_code}') # 1. 获取数据 df_raw = fetch_stock_daily(stock_code, start_date=start_date) if df_raw is None: return # 2. 计算基础特征 df = calculate_basic_features(df_raw) # 3. 计算动量指标 df = calculate_macd(df) df = calculate_rsi(df) df = identify_macd_signals(df) df = identify_rsi_signals(df) # 4. 计算成交量指标 df = calculate_obv(df) df = analyze_volume_price(df) # 5. 趋势与支撑阻力分析 df, support_levels, resistance_levels = identify_support_resistance(df) df = check_trend_strength(df) # 6. 生成综合信号(示例逻辑) # 潜在反弹信号组合:价格接近支撑位 + RSI超卖或金叉 + 成交量萎缩后温和放量 df['potential_bounce_signal'] = ( # 条件1: 收盘价接近最近识别的一个支撑位(在3%以内) (df['close'] <= min(support_levels) * 1.03 if len(support_levels) > 0 else False) & # 条件2: RSI出现看涨信号(从超卖区回升或低位金叉) (df['rsi_bullish_exit'] | ((df['rsi'] < 45) & df['macd_golden_cross'])) & # 条件3: 成交量出现止跌企稳迹象(当前不是放量下跌) (~df['volume_down_with_price']) ).shift(1) # 使用前一天的信号 print(f"识别到的潜在反弹信号日期: {df[df['potential_bounce_signal']].index.tolist()}") return df, support_levels, resistance_levels def plot_analysis_result(df, stock_name='东山精密'): """ 绘制综合分析图表 """ # 创建子图 fig = plt.figure(figsize=(16, 12)) # 子图1: K线图 + 均线 + 布林带 + 支撑阻力位 ax1 = plt.subplot(4, 1, 1) ax1.set_title(f'{stock_name} - 价格与趋势分析') # 这里可以绘制K线,为简化先绘制收盘价和均线 ax1.plot(df.index, df['close'], label='Close', linewidth=1, color='black') ax1.plot(df.index, df['sma_20'], label='SMA 20', linewidth=1, alpha=0.7) ax1.plot(df.index, df['bb_upper'], label='BB Upper', linestyle='--', linewidth=0.8, alpha=0.6) ax1.plot(df.index, df['bb_lower'], label='BB Lower', linestyle='--', linewidth=0.8, alpha=0.6) ax1.fill_between(df.index, df['bb_upper'], df['bb_lower'], alpha=0.1) # 标记信号点 signal_dates = df[df['potential_bounce_signal']].index ax1.scatter(signal_dates, df.loc[signal_dates, 'close'], color='red', s=50, zorder=5, label='Potential Bounce Signal') ax1.legend() ax1.grid(True, alpha=0.3) # 子图2: MACD ax2 = plt.subplot(4, 1, 2, sharex=ax1) ax2.set_title('MACD') ax2.plot(df.index, df['macd'], label='MACD', linewidth=1) ax2.plot(df.index, df['macd_signal'], label='Signal', linewidth=1) ax2.bar(df.index, df['macd_diff'], label='Histogram', alpha=0.5) ax2.axhline(y=0, color='grey', linestyle='-', linewidth=0.5) ax2.legend() ax2.grid(True, alpha=0.3) # 子图3: RSI ax3 = plt.subplot(4, 1, 3, sharex=ax1) ax3.set_title('RSI (14)') ax3.plot(df.index, df['rsi'], label='RSI', linewidth=1.5, color='purple') ax3.axhline(y=70, color='red', linestyle='--', linewidth=0.8, alpha=0.7, label='Overbought (70)') ax3.axhline(y=30, color='green', linestyle='--', linewidth=0.8, alpha=0.7, label='Oversold (30)') ax3.fill_between(df.index, 30, 70, alpha=0.1) ax3.legend() ax3.grid(True, alpha=0.3) # 子图4: 成交量 ax4 = plt.subplot(4, 1, 4, sharex=ax1) ax4.set_title('Volume') # 用颜色区分上涨日和下跌日 colors = ['red' if x < 0 else 'green' for x in df['returns']] ax4.bar(df.index, df['volume'], color=colors, alpha=0.6, label='Volume') ax4.plot(df.index, df['volume'].rolling(20).mean(), color='blue', linewidth=1.5, label='Volume MA 20') ax4.legend() ax4.grid(True, alpha=0.3) plt.tight_layout() plt.savefig(f'{stock_name}_technical_analysis.png', dpi=150) plt.show() if __name__ == '__main__': # 分析东山精密 df_result, supports, resistances = comprehensive_analysis('002384', '20240101') if df_result is not None: print(f"识别到的支撑位 (近5个): {supports}") print(f"识别到的阻力位 (近5个): {resistances}") plot_analysis_result(df_result.tail(100), '东山精密') # 绘制最近100天数据

运行以上代码,你将得到一份包含价格、MACD、RSI和成交量的综合分析图表,并标记出根据我们设定的复合条件识别出的“潜在反弹信号”点。这模拟了技术分析者寻找买点的一个逻辑闭环:价格到达关键支撑、动量指标出现背离或反转信号、成交量配合。

6. 常见问题与排查思路

在实践技术分析时,会遇到各种数据和代码问题。

问题现象常见原因解决思路
akshare获取数据返回空或报错1. 股票代码格式错误(需用数字代码)
2. 网络问题或接口暂时失效
3. 日期范围无数据(停牌等)
1. 确认代码格式,如'002384'
2. 检查网络,重试,或使用try-except捕获异常
3. 调整日期范围,或检查股票是否在该时段上市
指标计算出现NaN值滚动窗口(如rolling(20))在数据前19条无法计算使用.dropna()或.fillna(method='bfill')处理,或在分析时忽略开头的NaN
图表显示异常或错位1. 数据索引不是DatetimeIndex
2. 不同子图数据长度不一致
1. 确保使用df.set_index('date', inplace=True)
2. 使用sharex参数共享x轴,并确保绘图前数据已按时间排序
信号过多或过少信号触发条件过于宽松或苛刻调整信号组合的逻辑条件(如RSI阈值、均线周期、成交量倍数),并进行历史回测评估效果
分析结果与实盘感觉不符1. 技术指标滞后性
2. 未考虑大盘环境、板块效应
3. 忽略重大消息面影响
1. 承认技术分析的局限性,结合多时间周期分析
2. 加入板块指数(如半导体指数)作为参照
3. 技术分析需与基本面、市场情绪结合使用

7. 工程实践与高级策略建议

将上述分析流程工程化,并思考更稳健的策略。

7.1 构建可复用的分析流水线

# pipeline/analysis_pipeline.py from abc import ABC, abstractmethod import pandas as pd class AnalysisStep(ABC): """分析步骤的抽象基类""" @abstractmethod def process(self, df): pass class DataFetcherStep(AnalysisStep): def __init__(self, stock_code, **kwargs): self.stock_code = stock_code self.kwargs = kwargs def process(self, df=None): from utils.data_fetcher import fetch_stock_daily return fetch_stock_daily(self.stock_code, **self.kwargs) class FeatureEngineerStep(AnalysisStep): def process(self, df): from utils.data_processor import calculate_basic_features return calculate_basic_features(df) class MomentumAnalysisStep(AnalysisStep): def process(self, df): from indicators.momentum_indicators import calculate_macd, calculate_rsi df = calculate_macd(df) df = calculate_rsi(df) return df class AnalysisPipeline: def __init__(self): self.steps = [] def add_step(self, step): self.steps.append(step) def run(self): df = None for step in self.steps: df = step.process(df) return df # 使用示例 pipeline = AnalysisPipeline() pipeline.add_step(DataFetcherStep('002384', start_date='20240101')) pipeline.add_step(FeatureEngineerStep()) pipeline.add_step(MomentumAnalysisStep()) result_df = pipeline.run()

7.2 多维度验证与风控

单一技术信号不可靠,必须多维度验证:

  1. 多时间周期共振:日线出现买点,同时检查周线是否也处于支撑位或趋势好转。
  2. 板块与大盘确认:个股信号出现时,所属行业板块指数是否同步企稳?大盘指数(如沪深300)环境如何?
  3. 信号强度过滤:为信号设置“强度”分数。例如,同时满足“支撑位+RSI底背离+成交量倍缩”的信号比单一“RSI超卖”信号更强。
  4. 设置明确的止损位:任何基于技术分析的交易都必须有止损计划。例如,将止损设在支撑位下方3%-5%。

7.3 关于“汇金”等权重股的分析差异

对于“汇金股份”或“中国平安”等大盘权重股,技术分析逻辑相同,但需注意:

  • 趋势性更强:受单一资金推动的影响小于小盘股,趋势一旦形成,惯性更大。
  • 指标参数调整:波动率可能较低,布林带宽度、ATR等参数感受不同。
  • 需更重视市场整体流动性:其走势与大盘指数(上证50、沪深300)关联度极高,分析时务必结合指数状态。

技术分析是概率的游戏,是帮助决策的工具,而非预测未来的水晶球。本文提供的代码框架,旨在将主观的“逻辑分析”转化为客观的、可回溯检验的数据流程。真正的“达人逻辑”往往是多重信号过滤 + 严格风控管理 + 对市场情绪的深刻理解的综合体。建议读者在理解代码的基础上,不断调整参数、增加新的指标(如KDJ, DMI, 筹码分布等)、进行历史回测,从而形成适合自己的分析体系。市场有风险,任何分析都需谨慎对待。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询