企业官网建设流程全解析

最近有不少朋友在问，能不能把"看盘"这件事变成可复现、可回测的代码流程——不是那种玄学画线，而是把价格行为（Price Action）里的结构、形态、量能拆成能算的数值。这篇文章就把这套东西从零搭一遍：数据怎么拿、特征怎么算、信号怎么打分、回测怎么验证、哪些坑必须提前躲。全文用 Python 实现，代码分模块给出，拿到就能跑，跑完就能改。

需要先把话说在前面：本文所有内容都是技术实现层面的讨论，目的是演示如何用工程手段处理行情数据、如何做特征工程和回测框架。文中出现的任何信号、形态、打分，都不构成投资建议，也不能作为交易依据。市场价格受多重因素影响，历史规律不代表未来表现。涉及真实资金的操作务必谨慎，任何策略都应该先在模拟环境中充分验证。

本文适合三类读者：一是有一定 Python 基础，想入门量化分析方向的同学；二是做过数据处理，但没接触过行情数据结构、不知道 K 线怎么转成特征的后端开发者；三是有交易经验，想把"盘感"沉淀成代码逻辑，方便复盘和检索的实践者。读完之后，你应该能独立完成一套从数据加载、特征计算、形态识别、信号打分到简易回测的完整链路，并且知道每一环最容易出问题的地方在哪。

1. 背景与核心概念

1.1 价格行为分析到底在分析什么

价格行为分析，英文 Price Action，直译就是"价格的动作"。它和传统技术指标最大的区别在于：它不看 MACD 有没有金叉，不看 KDJ 是不是超卖，而是直接盯着最原始的四个数字——开盘价、最高价、最低价、收盘价，再加上成交量。

为什么这件事值得单独拿出来做？因为所有技术指标本质上都是这四个价格经过某种公式变换后的产物。均线是收盘价的移动平均，MACD 是两条均线的差再平滑，RSI 是涨跌幅的比值。这些指标在变换过程中一定会丢失信息，而且会引入滞后。价格行为分析的思路是回到源头：既然指标是从价格推出来的，那我为什么不直接研究价格本身的结构？

举个具体的例子。一根 K 线，如果实体很小、下影线很长、并且出现在一段下跌之后，传统指标可能还在显示"下跌趋势中"，但价格行为会把这一根 K 线识别为"下方有承接力量"的信号。这个判断不依赖任何公式参数，只依赖 K 线本身的形状和它所处的位置。

所以价格行为分析的核心构件其实是三块：

第一块是K 线形态。单根 K 线的实体与影线比例（pin bar、十字星）、两根 K 线的包含关系（inside bar、outside bar）、多根 K 线的组合（双顶双底、旗形整理）。

第二块是市场结构。价格在上涨过程中形成的"更高的高点 + 更高的低点"，在下跌过程中形成的"更低的高点 + 更低的低点"，以及结构被破坏的那个瞬间。这是判断趋势状态最朴素也最有效的方法。

第三块是关键价位。历史上反复被测试的支撑位和阻力位、前高前低、整数关口、成交密集区。这些位置之所以有效，一部分原因是订单确实堆积在那里，另一部分原因是大量参与者都在看同样的位置。

把这三块拆开看，你会发现每一块其实都是可以量化的。这就是本文要做的事情。

1.2 为什么必须用代码来做

手工画线做价格行为分析，一天复盘三五十只股票就已经到极限了。而且人有个天然的毛病：记忆有偏。昨天判断错的形态，今天大概率想不起来；昨天判断对的形态，会被反复强化。这种选择性记忆会让"盘感"越来越不准，但你本人是察觉不到的。

代码化的第一个价值是一致性。同一套规则，今天跑和三个月后跑，结果必须一样。如果不一样，那说明代码里有随机性或者数据有变化，这本身就是一个需要排查的信号。

代码化的第二个价值是可回测。你可以把过去三年的数据全部跑一遍，统计某个形态出现之后，未来 1 天、3 天、5 天、10 天的收益分布是什么样的，胜率多少、盈亏比多少、最大回撤多少。这些数字是手工复盘永远算不出来的。

代码化的第三个价值是可扩展。当你把 K 线转成了特征表，后面接规则引擎也行，接机器学习模型也行，接大模型做文本解读也行，底座是同一套。不用每次换方法就重写一遍数据层。

但同时必须清楚两个边界。第一个边界是：代码只能处理"能被明确定义的东西"。像"市场情绪偏弱"这种描述，如果你不能把它翻译成成交量的相对倍数、涨跌家数比、振幅变化率，那它就没法进代码。第二个边界是：回测结果好，不代表实盘结果好。回测里没有滑点、没有流动性约束、没有涨跌停无法成交的情况。这两点在后面回测那一节会专门展开。

2. 环境准备与项目结构

2.1 运行环境与依赖说明

本文示例使用 Python 3.9 及以上版本，理论上 3.8 也能跑，但部分语法糖会受限。操作系统 Windows、macOS、Linux 都可以，代码本身不涉及平台相关调用。

核心依赖只有四个：

pip install pandas numpy akshare matplotlib

简单说明每个的作用。pandas负责表格型数据的组织和计算，是整条链路的地基；numpy提供向量化运算，在计算滚动极值、条件判断时能显著提速；akshare是一个开源的金融数据接口库，用来获取 A 股历史行情；matplotlib只在需要画图验证的时候用，可选。

这里要特别提醒一句：具体版本号请以你本地实际安装的为准，本文不锁定版本。数据接口类库的字段名、参数名在版本迭代中确实发生过变化，如果你装的是较新的版本，返回的列名可能和文中不完全一致，这时候用print(df.columns)打印一下，按实际字段做映射即可。这个习惯比记住某个具体版本号有用得多。

如果你的网络环境访问数据接口不稳定，也可以完全跳过akshare，用本地 CSV 文件作为数据源。本文的数据加载模块设计了两种模式，CSV 模式只需要保证列名符合约定即可。

2.2 项目目录结构设计

pa_analyzer/ ├── data_loader.py # 数据加载与清洗 ├── features.py # 特征工程 ├── patterns.py # 形态识别 ├── engine.py # 信号打分引擎 ├── backtest.py # 简易回测 ├── report.py # 结果输出与可视化 └── main.py # 主流程入口

这个结构不是随便定的。分成六层的原因在于：每一层的输入输出边界清晰，单独替换不会影响其他层。比如你后面想把akshare换成别家的数据源，只改data_loader.py，其他五个文件一行都不用动。再比如你想把规则引擎换成机器学习模型，只改engine.py，前面的特征计算照常复用。

很多初学者习惯把所有逻辑写在一个main.py里，几百行下来改一个参数要找半天，而且没法单测。分层的成本是多建几个文件，收益是后面每一次修改都省时间。

2.3 数据源的合规提醒

使用任何行情数据之前，请确认数据来源的使用条款。公开接口提供的数据通常有频率限制，用于个人学习和研究一般没问题，但如果要做批量抓取、对外提供服务或者商业用途，需要单独确认授权。本文示例控制在小批量、低频次的范围内，符合常见的使用场景。

另外，A 股有它自己的一套规则，写代码时必须考虑进去，否则回测结果会严重失真：

• 涨跌停制度。主板股票日涨跌幅限制为 10%，创业板和科创板为 20%，ST 类股票为 5%。如果信号触发当天股票一字涨停，你是买不进去的，回测时必须把这个情况排除。
• T+1 制度。当天买入的股票当天不能卖出，最短持有期是 1 个交易日。回测里如果按"当天收盘买入、当天收盘卖出"来算，结果完全是假的。
• 停牌。停牌期间没有行情数据，如果你按日期索引对齐，会出现空缺。处理方式要么是前向填充（但价格不能填），要么是在信号判断时跳过。

这三点是 A 股回测最常见的失真来源，比参数调优重要得多。

3. 核心原理拆解：把"盘感"翻译成可计算的量

3.1 K 线结构的三个基础量

一根 K 线有四个价格，能衍生出几个非常有用的比例量。这些量构成了后面所有形态识别的基础。

实体占比：实体长度除以整根 K 线的总长度。实体长度是abs(close - open)，总长度是high - low。实体占比接近 1，说明多空一方完全主导；接近 0，说明双方拉锯，收盘价回到了开盘价附近。

上影线占比：high - max(open, close)除以总长度。上影线长，说明价格冲高之后被打回来了，上方有卖压。

下影线占比：min(open, close) - low除以总长度。下影线长，说明价格下探之后被买回来了，下方有承接。

振幅：(high - low) / 前收盘价。这个量衡量的是当天的波动剧烈程度，和形态无关，但对过滤无效信号很有用。振幅太小（比如小于 1%）的 K 线，形态本身没有意义，因为它只是正常的波动噪音。

这里有个容易踩的坑：当high == low（一字板）的时候，总长度为 0，直接做除法会得到inf或者NaN。代码里必须做保护，通常的做法是把总长度为 0 的 K 线的各个占比直接置为 0 或者NaN，然后在后续判断中过滤掉。

3.2 摆动高低点与趋势状态机

"更高的高点、更高的低点"这句话，人眼一眼就能看出来，但怎么让程序识别？关键在于先把"摆动高低点"（swing high / swing low）找出来。

摆动高点的定义是：某根 K 线的最高价，比它左边 N 根和右边 N 根的最高价都高。这个 N 通常取 2 到 3。参数取太小，会找出大量噪音点；取太大会漏掉短周期的结构转折。

找到摆动点之后，趋势判断就变成了一个状态机：

• 如果最近一个摆动高点比上一个摆动高点高，并且最近一个摆动低点也比上一个摆动低点高，状态是"上升结构"。
• 如果最近一个摆动高点比上一个低，并且最近一个摆动低点也比上一个低，状态是"下降结构"。
• 其他情况（高低点方向不一致），状态是"震荡"。

这个状态机有个重要特性：它是滞后确认的。也就是说，一个摆动点必须等右边 N 根 K 线走完才能被确认。这意味着在实盘中，你永远无法在摆动点形成的那一根 K 线上就知道它是摆动点。这个滞后是结构判断的固有成本，无法消除，只能接受。很多"回测很神、实盘很差"的策略，问题就出在这里——回测时用了未来数据。

3.3 支撑阻力区的量化

支撑阻力不能当成一条精确的线来处理，因为价格从来不会精确到分地尊重某条线。正确做法是把它处理成一个区间。

一种简单有效的做法是：取出过去一段时间内所有的摆动高点和摆动低点，把这些价格按照当前股价的百分比（比如 1%）分桶，统计每个桶里落了多少个摆动点。落得多的桶，说明这个价格区域被反复测试，就是强支撑或强阻力。

举个例子，某只股票当前价格 20 元，1% 的分桶宽度是 0.2 元。过去半年里，19.8 到 20.0 这个区间出现了 5 次摆动点，19.0 到 19.2 出现了 4 次，那 19.8 到 20.0 就是一个更强的关键位。

这个方法的好处是完全由数据驱动，不需要人工画线。缺点是它假设价格分布是静态的，如果公司基本面发生重大变化，历史的支撑位可能已经失效。所以实际使用中，通常只取最近 120 到 250 个交易日的数据来做聚类，太久远的数据参考价值有限。

3.4 成交量的相对量能

成交量的绝对值没有意义，贵州茅台的日成交量和小盘股的日成交量差了好几个数量级，直接比较毫无价值。真正有意义的是相对量能：今天的成交量是过去 20 天平均成交量的多少倍。

量比 = 今日成交量 / 过去 20 日成交量均值

量比大于 1.5 通常被视为放量，小于 0.7 视为缩量。这个阈值的具体数值需要根据标的的波动特征调整，没有通用标准。

放量在不同位置的意义完全不同，这一点必须写进代码逻辑里：

• 突破前高时放量，说明有真实资金在推动，突破的可信度高。
• 突破前高时缩量，说明只是少量资金在拉，很可能是假突破。
• 下跌末端放量，可能是恐慌盘出清，也可能是资金出逃，需要结合 K 线形态判断。
• 下跌末端缩量，说明抛压衰竭，往往是筑底信号。

所以量能不能单独作为信号，必须和价格结构结合。这也是为什么后面的打分引擎里，量能只作为加分项，不单独触发信号。

4. 特征工程与形态识别的代码实现

4.1 数据加载模块

先写数据层。这个模块的职责很单一：把原始数据变成一张干净的、列名统一的 DataFrame。

# 文件路径：pa_analyzer/data_loader.py import pandas as pd STANDARD_COLUMNS = ["date", "open", "high", "low", "close", "volume"] def load_from_akshare(symbol: str, start: str, end: str) -> pd.DataFrame: """从开源接口加载 A 股日线数据，前复权。 symbol: 6 位股票代码，例如 "600519" start / end: 形如 "20210101" """ import akshare as ak raw = ak.stock_zh_a_hist( symbol=symbol, period="daily", start_date=start, end_date=end, adjust="qfq", ) if raw is None or raw.empty: raise ValueError(f"未获取到 {symbol} 的数据，请检查代码或日期区间") column_map = { "日期": "date", "开盘": "open", "收盘": "close", "最高": "high", "最低": "low", "成交量": "volume", } # 只映射存在的列，避免因版本差异导致的 KeyError raw = raw.rename(columns={k: v for k, v in column_map.items() if k in raw.columns}) missing = [c for c in STANDARD_COLUMNS if c not in raw.columns] if missing: raise ValueError(f"数据缺少必要列: {missing}，实际列为: {list(raw.columns)}") df = raw[STANDARD_COLUMNS].copy() df["date"] = pd.to_datetime(df["date"]) df = df.sort_values("date").drop_duplicates(subset="date") df = df.reset_index(drop=True) return df def load_from_csv(path: str) -> pd.DataFrame: """从本地 CSV 加载，列名需包含 date/open/high/low/close/volume。""" df = pd.read_csv(path) missing = [c for c in STANDARD_COLUMNS if c not in df.columns] if missing: raise ValueError(f"CSV 缺少必要列: {missing}") df = df[STANDARD_COLUMNS].copy() df["date"] = pd.to_datetime(df["date"]) df = df.sort_values("date").drop_duplicates(subset="date").reset_index(drop=True) return df

这里有两个细节值得说明。第一个是rename时的字典推导式过滤：直接写raw.rename(columns=column_map)的话，如果新版本接口改了列名，某些映射会失效，但不会报错，你会拿到一堆莫名其妙的NaN。加上过滤并做missing检查之后，问题会在第一时间暴露出来。第二个是drop_duplicates(subset="date")，有些数据源在除权除息日附近会出现重复日期，不去重会导致后续滚动计算全部错位。

4.2 基础特征计算

# 文件路径：pa_analyzer/features.py import numpy as np import pandas as pd def add_basic_features(df: pd.DataFrame) -> pd.DataFrame: df = df.copy() rng = (df["high"] - df["low"]).replace(0, np.nan) body = (df["close"] - df["open"]).abs() upper_shadow = df["high"] - df[["open", "close"]].max(axis=1) lower_shadow = df[["open", "close"]].min(axis=1) - df["low"] df["body_ratio"] = (body / rng).fillna(0.0) df["upper_ratio"] = (upper_shadow / rng).fillna(0.0) df["lower_ratio"] = (lower_shadow / rng).fillna(0.0) df["is_bull"] = (df["close"] > df["open"]).astype(int) # 真实波幅 TR 与 ATR，用于做波动率归一化 prev_close = df["close"].shift(1) tr = pd.concat( [ df["high"] - df["low"], (df["high"] - prev_close).abs(), (df["low"] - prev_close).abs(), ], axis=1, ).max(axis=1) df["tr"] = tr df["atr"] = tr.ewm(alpha=1 / 14, adjust=False).mean() # 相对量能 df["vol_ma20"] = df["volume"].rolling(20, min_periods=5).mean() df["vol_ratio"] = (df["volume"] / df["vol_ma20"]).round(3) return df

这里atr用的是 Wilder 平滑，等价于alpha = 1/14的指数移动平均。为什么需要 ATR？因为后面判断"这根 K 线的波动是否够大"时，如果用百分比绝对值，不同股票的阈值完全不一样。用振幅 / ATR做归一化之后，所有标的就统一了尺度。这是量化里非常常用的手法，也叫波动率标准化。

4.3 摆动点识别与趋势标注

# 文件路径：pa_analyzer/features.py（续） def add_swing_points(df: pd.DataFrame, left: int = 2, right: int = 2) -> pd.DataFrame: df = df.copy() highs = df["high"].to_numpy() lows = df["low"].to_numpy() n = len(df) swing_high = np.zeros(n, dtype=bool) swing_low = np.zeros(n, dtype=bool) for i in range(left, n - right): window_high = highs[i - left: i + right + 1] # 必须严格大于左侧、大于等于右侧，避免平台期重复标记 if highs[i] == window_high.max() and highs[i] > window_high[:left].max(): swing_high[i] = True window_low = lows[i - left: i + right + 1] if lows[i] == window_low.min() and lows[i] < window_low[:left].min(): swing_low[i] = True df["swing_high"] = swing_high df["swing_low"] = swing_low return df def add_trend_state(df: pd.DataFrame) -> pd.DataFrame: df = df.copy() trend = [] prev_sh = last_sh = None prev_sl = last_sl = None current = "range" for _, row in df.iterrows(): if row["swing_high"]: prev_sh, last_sh = last_sh, row["high"] if row["swing_low"]: prev_sl, last_sl = last_sl, row["low"] if prev_sh is not None and prev_sl is not None: if last_sh > prev_sh and last_sl > prev_sl: current = "up" elif last_sh < prev_sh and last_sl < prev_sl: current = "down" else: current = "range" trend.append(current) df["trend"] = trend return df

关于摆动点的判定条件，我加了highs[i] > window_high[:left].max()这一层限制，也就是要求它严格大于左侧。原因是不加限制的话，一段平台整理会被标记成一连串的摆动高点，趋势状态机就会被这些噪音点反复拉扯，输出一堆无意义的翻转。这类细节在教科书里很少提，但在实际跑数据时影响非常大。

add_trend_state里面用了iterrows，这在数据量大时性能一般。如果要做全市场扫描，建议改成先取出摆动点的索引数组，再用循环处理索引，会快不少。为了让代码可读性优先，这里保留了逐行遍历的写法。

4.4 形态识别

# 文件路径：pa_analyzer/patterns.py import numpy as np import pandas as pd def detect_pin_bar(df: pd.DataFrame, shadow_ratio: float = 0.6, body_limit: float = 0.3) -> pd.DataFrame: """识别锤子线 / 上吊线风格的长影线 K 线。""" df = df.copy() # 下影线长、实体小 -> 下方有承接 df["pin_bull"] = ( (df["lower_ratio"] >= shadow_ratio) & (df["body_ratio"] <= body_limit) & (df["upper_ratio"] <= 0.25) ).astype(int) # 上影线长、实体小 -> 上方有抛压 df["pin_bear"] = ( (df["upper_ratio"] >= shadow_ratio) & (df["body_ratio"] <= body_limit) & (df["lower_ratio"] <= 0.25) ).astype(int) return df def detect_inside_outside(df: pd.DataFrame) -> pd.DataFrame: df = df.copy() prev_high = df["high"].shift(1) prev_low = df["low"].shift(1) df["inside_bar"] = ((df["high"] <= prev_high) & (df["low"] >= prev_low)).astype(int) df["outside_bar"] = ((df["high"] > prev_high) & (df["low"] < prev_low)).astype(int) return df def detect_breakout(df: pd.DataFrame, window: int = 20) -> pd.DataFrame: df = df.copy() # 使用 shift(1) 保证只用到今天之前的数据，避免未来函数 prior_high = df["high"].rolling(window).max().shift(1) prior_low = df["low"].rolling(window).min().shift(1) df["break_up"] = (df["close"] > prior_high).astype(int) df["break_down"] = (df["close"] < prior_low).astype(int) df["prior_high_20"] = prior_high df["prior_low_20"] = prior_low return df

这里再次强调shift(1)的重要性。如果不加这个，rolling(window).max()会包含当天的最高价，而当天最高价一定大于等于当天收盘价，所以close > prior_high这个条件永远不成立，突破信号一个都检测不出来。这种错误叫"未来函数"，是回测里最常见也最隐蔽的 bug 之一，因为程序不报错，只是结果完全不对。

5. 信号打分引擎的设计

5.1 为什么用打分而不是布尔判断

很多教程的做法是：满足条件 A 并且满足条件 B 就发出信号。这种写法的问题是太刚性。比如"趋势向上"这个条件，如果某天刚好结构从一个明确的上升转成震荡，信号就直接消失了，但实际上那天的形态可能非常优质。

打分制的思路是：信号强度 = 各个维度得分的加权和。每个维度独立判断，得分可以叠加，最终根据总分划分等级。这样做的好处是能容忍部分条件不满足，同时也能输出"信号强度"这个连续量，方便后续做阈值调优。

# 文件路径：pa_analyzer/engine.py import pandas as pd def score_signals(df: pd.DataFrame, cfg: dict = None) -> pd.DataFrame: cfg = cfg or { "w_trend": 25, "w_pattern": 30, "w_volume": 20, "w_break": 25, "min_score": 60, } df = df.copy() scores, tags = [], [] for _, row in df.iterrows(): score = 0 tag = [] # 维度一：趋势结构 if row["trend"] == "up": score += cfg["w_trend"] tag.append("上升结构") elif row["trend"] == "down": score -= cfg["w_trend"] tag.append("下降结构") # 维度二：K 线形态 if row["pin_bull"] == 1: score += cfg["w_pattern"] tag.append("长下影承接") if row["pin_bear"] == 1: score -= cfg["w_pattern"] tag.append("长上影抛压") if row["inside_bar"] == 1: tag.append("内包整理") # 维度三：量能配合 if pd.notna(row["vol_ratio"]): if row["vol_ratio"] >= 1.5: score += cfg["w_volume"] tag.append("放量") elif row["vol_ratio"] <= 0.7: score -= cfg["w_volume"] * 0.5 tag.append("缩量") # 维度四：关键位突破 if row["break_up"] == 1: score += cfg["w_break"] tag.append("向上突破") if row["break_down"] == 1: score -= cfg["w_break"] tag.append("向下破位") # 振幅过滤：波动太小的 K 线形态可信度低 if pd.notna(row["atr"]) and row["atr"] > 0: if (row["high"] - row["low"]) < 0.3 * row["atr"]: score = int(score * 0.7) tag.append("振幅过小降权") scores.append(score) tags.append("+".join(tag) if tag else "") df["score"] = scores df["tag"] = tags df["signal"] = df["score"].apply( lambda s: "看多" if s >= cfg["min_score"] else ("看空" if s <= -cfg["min_score"] else "观望") ) return df

这段代码有几个设计意图需要解释。

第一，权重是可以配置的。cfg字典把四个维度的权重和触发阈值全部外置了。这意味着你可以在不改代码的前提下，通过传入不同配置来做参数对比实验。参数硬编码在函数里是维护噩梦，一旦要测十组参数，你就得改十次代码。

第二，正向和负向是对称的。看多的条件加分，看空的条件减分，最后用正负分判断方向。这样做的好处是一套代码同时处理多空，不用写两份逻辑。

第三，振幅降权是一个软过滤。与其直接丢弃信号，不如把分数打折。这样既降低了噪音信号的影响，又保留了记录，方便后续复盘时看到"这里本来有个信号但被我打折了"。

5.2 从规则引擎升级到机器学习模型

规则引擎的好处是透明、可解释、可调。缺点是阈值全靠人工设定，而且各维度之间是线性叠加，捕捉不到交互效应。

如果要把这套特征接机器学习，思路其实很直接：特征表已经有了，只需要构造标签。典型做法是定义未来 N 日收益 > 阈值为正样本，然后训练一个二分类模型。

# 文件路径：pa_analyzer/engine.py（续） import numpy as np from sklearn.ensemble import GradientBoostingClassifier def build_ml_dataset(df: pd.DataFrame, horizon: int = 5, threshold: float = 0.02): data = df.copy() data["fwd_ret"] = data["close"].shift(-horizon) / data["close"] - 1 data = data.dropna(subset=["fwd_ret"]) feature_cols = [ "body_ratio", "upper_ratio", "lower_ratio", "vol_ratio", "pin_bull", "pin_bear", "inside_bar", "break_up", "break_down", ] data["label"] = (data["fwd_ret"] > threshold).astype(int) X = data[feature_cols].fillna(0) y = data["label"] return X, y, feature_cols

这里必须提醒一个极其重要的问题：绝对不能随机打乱做 train_test_split。时间序列数据有严格的先后顺序，随机切分会让模型在训练时看到未来的数据，评估结果会好得离谱，但实盘完全没用。正确做法是按时间切分，前 70% 训练，后 30% 测试，或者用滚动窗口做 walk-forward 验证。

另外，label的阈值 0.02 只是示例。这个值要和你的持有周期匹配，5 天涨 2% 和 30 天涨 2% 完全是两回事。同时要注意正负样本比例，如果正样本只占 15%，模型的准确率指标会失去意义，应该看 AUC 或者精确率-召回率曲线。

6. 完整实战：从拉数据到出报告

6.1 安装依赖并准备目录

mkdir pa_analyzer && cd pa_analyzer pip install pandas numpy akshare matplotlib scikit-learn

创建前面列出的六个 Python 文件，把代码分别填进去。

6.2 编写主流程

# 文件路径：pa_analyzer/main.py import pandas as pd from data_loader import load_from_akshare from features import add_basic_features, add_swing_points, add_trend_state from patterns import detect_pin_bar, detect_inside_outside, detect_breakout from engine import score_signals def analyze(symbol: str, start: str, end: str) -> pd.DataFrame: df = load_from_akshare(symbol, start, end) df = add_basic_features(df) df = add_swing_points(df, left=2, right=2) df = add_trend_state(df) df = detect_pin_bar(df) df = detect_inside_outside(df) df = detect_breakout(df, window=20) df = score_signals(df) return df if __name__ == "__main__": result = analyze("600519", "20220101", "20241231") cols = ["date", "close", "trend", "score", "signal", "tag"] print(result[cols].tail(20).to_string(index=False)) hot = result[result["signal"] != "观望"] print(f"\n统计区间共 {len(result)} 个交易日，触发信号 {len(hot)} 次") print(hot["signal"].value_counts()) result.to_csv("result_600519.csv", index=False, encoding="utf-8-sig")

6.3 运行与结果解读

执行python main.py，你会看到两部分输出。第一部分是最近 20 个交易日的明细表，包含日期、收盘价、趋势状态、综合打分、信号结论和触发原因。第二部分是区间内的信号统计。

预期输出大致长这样（具体数字取决于实际数据）：

date close trend score signal tag 2024-12-02 1523.00 up 80 看多 上升结构+放量+向上突破 2024-12-03 1541.50 up 30 观望 上升结构 ...

重点看tag这一列。它把你触发信号的每一个理由都列出来了，这是这套系统相比黑盒模型最大的优势——你能立刻知道"为什么今天给出看多"。如果某天你觉得这个信号不合理，直接看 tag 就知道是哪个维度误判了，然后去调对应的阈值。

6.4 加一个简单的可视化

# 文件路径：pa_analyzer/report.py import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt def plot_signals(df, save_path="signals.png", last_n=120): d = df.tail(last_n) fig, (ax1, ax2) = plt.subplots( 2, 1, figsize=(14, 8), sharex=True, gridspec_kw={"height_ratios": [3, 1]} ) ax1.plot(d["date"], d["close"], linewidth=1.2, label="close") ax1.plot(d["date"], d["prior_high_20"], linewidth=0.8, linestyle="--", label="20D high") bull = d[d["signal"] == "看多"] bear = d[d["signal"] == "看空"] ax1.scatter(bull["date"], bull["close"], marker="^", s=60, label="bull") ax1.scatter(bear["date"], bear["close"], marker="v", s=60, label="bear") ax1.legend(loc="best") ax1.set_title("Price Action Signals") ax2.bar(d["date"], d["vol_ratio"], width=1.0) ax2.axhline(1.5, linestyle="--", linewidth=0.8) ax2.axhline(0.7, linestyle="--", linewidth=0.8) ax2.set_ylabel("vol ratio") plt.tight_layout() plt.savefig(save_path, dpi=120) plt.close(fig)

用matplotlib.use("Agg")是为了在没有图形界面的服务器上也能出图，这个细节在本地跑可能感觉不到，但一旦上服务器就会报错。

7. 回测验证：把信号放到时间轴上检验

7.1 简易事件驱动回测

# 文件路径：pa_analyzer/backtest.py import pandas as pd def run_backtest(df: pd.DataFrame, horizon: int = 5, stop_atr: float = 1.5, target_atr: float = 3.0) -> dict: """信号次日开盘入场，持有 horizon 日或触发止损/止盈离场。 注意：此处未模拟涨跌停、滑点和手续费，结果仅供结构验证。 """ trades = [] n = len(df) for i in range(n - 1): if df.at[i, "signal"] != "看多": continue entry_idx = i + 1 entry = df.at[entry_idx, "open"] atr = df.at[i, "atr"] if pd.isna(atr) or atr <= 0: continue stop = entry - stop_atr * atr target = entry + target_atr * atr exit_price, exit_idx, reason = None, None, "" for j in range(entry_idx, min(entry_idx + horizon, n)): low, high = df.at[j, "low"], df.at[j, "high"] if low <= stop: exit_price, exit_idx, reason = stop, j, "stop" break if high >= target: exit_price, exit_idx, reason = target, j, "target" break if exit_price is None: exit_idx = min(entry_idx + horizon - 1, n - 1) exit_price = df.at[exit_idx, "close"] reason = "timeout" ret = exit_price / entry - 1 trades.append({ "entry_date": df.at[entry_idx, "date"], "exit_date": df.at[exit_idx, "date"], "entry": entry, "exit": exit_price, "ret": ret, "reason": reason, }) if not trades: return {"trades": [], "win_rate": 0.0, "avg_ret": 0.0, "count": 0} tdf = pd.DataFrame(trades) wins = (tdf["ret"] > 0).sum() return { "trades": trades, "count": len(tdf), "win_rate": round(wins / len(tdf), 4), "avg_ret": round(tdf["ret"].mean(), 4), "max_win": round(tdf["ret"].max(), 4), "max_loss": round(tdf["ret"].min(), 4), }

这段回测的框架是：信号出现当天不算，第二天开盘价入场，然后逐日检查是否触及止损或止盈，都没触及就按持有周期到期收盘价离场。这个逻辑比"信号当天收盘价买入"要真实得多，因为你在收盘时看到信号，实际上已经无法以那个价格成交了。

7.2 回测结果必须警惕的四个失真点

跑完之后你会拿到胜率、平均收益这些数字。但在看这些数字之前，先确认下面四件事有没有处理。

第一，涨跌停无法成交。如果信号次日开盘一字涨停，你根本买不进去。这个情况在强势股上非常常见，忽略它会让回测结果虚高。判断方法是检查次日open == high == low，或者涨跌幅 > 9.8%且open == high。

第二，滑点。实际成交价和你计算的价格之间会有差距，流动性越差的股票差距越大。保守做法是在买入价上加 0.1%，卖出价上减 0.1%。

第三，手续费。A 股目前买入有佣金（通常万分之一到万分之三，有最低 5 元限制），卖出还有印花税。高频策略里这部分成本占比很高。

第四，样本量。如果整个回测期只触发了 8 次信号，那 75% 的胜率毫无意义，统计上完全不显著。经验上，一个策略至少要有 30 次以上的独立信号，结论才有参考价值。

把这几条补上之后，很多看起来"胜率 70%"的策略会掉到 45% 以下。这不是代码写错了，而是之前的数字本来就是幻觉。

8. 常见问题与排查思路

这套代码跑起来不算复杂，但在实际使用中会遇到一些典型问题。下面按出现频率排列。

几个补充说明。关于"形态识别全为 0"，还有一种可能是你用的数据没有做复权处理。未复权的数据在除权日会出现巨大的价格跳空，导致high - low异常大，所有比例量都被稀释了，形态自然识别不出来。这个问题很隐蔽，因为数据本身是"正确"的，只是不适合做形态分析。

关于"分数出现 NaN"，vol_ratio在前面 20 天一定是NaN，因为它依赖 20 日均量。如果你直接拿NaN去比较大小，Python 会返回False而不报错，逻辑就悄悄地走偏了。所以代码里每次都显式判断pd.notna()。

关于"回测收益异常高"，这是最需要警惕的一类问题。经验上，如果一个简单规则策略回测出来年化收益超过 50% 且回撤很小，那几乎可以肯定是代码有问题，而不是你发现了什么秘密。这时候不要急着高兴，先把代码从头到尾查一遍数据泄露。

9. 最佳实践与工程建议

9.1 数据处理层面的约定

统一在数据层做复权。前复权适合做形态分析和回测，因为历史价格会被调整成和当前可比；后复权适合做长期收益计算。不要在不同模块混用两种口径，否则算出来的收益率会莫名其妙地对不上。

日期是主键，但日期可能重复或缺失。去重和排序一定要在加载后立刻做，不要留到后面。同时要注意，如果你按日期和另一个数据源做merge，一定要指定how="left"并用validate参数检查唯一性，否则行数会悄悄翻倍。

所有除法都要考虑分母为 0。这一点在前面的rng.replace(0, np.nan)里已经处理过了。同样的道理适用于vol_ratio计算，如果某天成交量为 0（停牌），vol_ma20会受影响，最好在计算前把停牌日剔除。

9.2 配置管理的正确姿势

前面score_signals的cfg参数是个好开头，但还停留在函数参数级别。更规范的做法是把所有可调参数抽到一个 YAML 文件里：

# config.yaml data: symbol: "600519" start: "20220101" end: "20241231" adjust: "qfq" pattern: swing_left: 2 swing_right: 2 pin_shadow_ratio: 0.6 breakout_window: 20 scoring: w_trend: 25 w_pattern: 30 w_volume: 20 w_break: 25 min_score: 60 backtest: horizon: 5 stop_atr: 1.5 target_atr: 3.0 commission: 0.0003 slippage: 0.001

配置文件的价值在于参数实验可追溯。你今天跑出来一组好参数，三个月后想复现，只要把配置文件找出来就行。如果参数散落在代码里，那些if ratio > 0.6你根本记不住当时为什么选 0.6。

9.3 异常处理与日志

数据类代码最容易出问题的地方不是逻辑，而是"数据不符合预期"。所以每个关键节点都应该有明确的失败信息：

def safe_analyze(symbol: str, start: str, end: str, logger=None): try: df = load_from_akshare(symbol, start, end) except Exception as e: if logger: logger.error(f"[{symbol}] 数据加载失败: {e}") return None if len(df) < 30: if logger: logger.warning(f"[{symbol}] 数据量不足 30 条，跳过分析") return None # ...后续流程 return df

不要写裸的except: pass。那会让所有问题静默失败，最后你只看到一个空结果，完全不知道是数据没拿到、还是计算报错、还是过滤条件太严。日志里带上股票代码和具体异常信息，排查效率会高一个数量级。

9.4 性能与可维护性

如果只分析几十只股票，现在的写法完全够用。但如果要做全市场 5000 只股票的扫描，性能就会成为瓶颈。这时候有几个明确的优化方向：

第一，特征计算向量化。add_basic_features已经全部是向量化操作，这部分没问题。瓶颈在add_swing_points和add_trend_state的逐行循环。前者可以改成用scipy.signal.argrelextrema，后者可以改成先提取摆动点索引数组再循环。

第二，缓存中间结果。数据加载往往是最慢的一环，可以把清洗后的 DataFrame 存成 parquet 文件，下次直接读，速度能快一个数量级。

第三，并发控制。抓取数据时不要开几十个线程猛冲，容易被限流，反而更慢。控制在 2 到 3 个并发，加上适当的间隔，整体效率反而更高。

9.5 生产环境的边界意识

最后必须强调一点：本文这套代码的定位是研究和复盘工具，不是自动交易系统。如果要在真实环境中使用，至少还需要补齐这些：实时的行情推送和断线重连、订单状态管理、异常熔断、仓位控制和风控规则、完整的审计日志。这些都是工程层面的硬需求，缺一个都可能出问题。

而且即使全部补齐，历史规律也不保证未来重现。市场结构会变，参与者会变，监管规则会变。任何基于历史数据的模型，都必须持续监控其有效性，一旦发现表现持续偏离预期，就要重新审视假设是否还成立。

10. 总结与学习路线

回过头看，这套系统其实只做了三件事：把 K 线翻译成结构化的数值特征，把交易经验翻译成可调权重的打分规则，把结果放到时间轴上做可复现的验证。三件事里，最容易出彩的是特征设计，最容易出错的是回测验证，最有长期价值的是分层架构。

如果你想继续往下深入，建议按这个顺序推进。

第一阶段，把特征做扎实。现在的特征只有十几个，可以补充：缺口（跳空）的识别与回补统计、连续 K 线的实体累积、波动率的变化率（ATR 的斜率）、成交量的分布形态（比如大单占比，如果有分笔数据的话）。特征越贴近你真实观察市场的方式，后面的模型就越有意义。

第二阶段，把验证做严格。现在的回测非常简陋，没有考虑涨跌停、滑点、手续费和仓位管理。可以先用backtrader或者vectorbt这类成熟框架替换掉手写的回测，重点是把交易成本模型建对。

第三阶段，把评估做全面。不要只看胜率和平均收益，要看最大回撤、收益分布、在不同市场状态（上涨、下跌、震荡）下的分段表现。一个策略如果只在单边上涨行情里赚钱，那它本质上是在赌方向，而不是在做价格行为。

至于机器学习那部分，建议放在最后再碰。规则引擎跑通、理解清楚每个特征的实际含义之后，再去接模型，你才知道模型给出的重要性排序是不是合理。反过来，一开始就上模型，你会陷入"调参调到天亮，却不知道哪里不对"的困境。

代码本身已经给全了，建议你拿自己熟悉的几只股票跑一遍，重点看的不是信号准不准，而是代码有没有按你预想的方式工作。把每个模块的中间结果打印出来对一遍，比看十篇教程都有用。真正跑起来、改起来、被报错折磨几轮之后，这套东西才算真正变成你自己的。