单只/少量有免费额度；全市场标的池扫描（`universes` + `klines.batch`）需 Starter 及以上。详见[定价页](https://alphafeed.org/)。

如何用 Python 筛选连续上涨（N 连阳）的股票？（全市场连阳/连阴统计）

Q: 怎么统计连阴（连续下跌）？

把判定条件反过来：`close < open`（连阴）或 `close < 昨收`（连跌），再用同样的 `current_streak`。

Q: 为什么要前复权？

除权日未复权会出现价格断层，可能把正常交易日误判成阴/阳或跳空，前复权可消除。

选股扫描 · 代码验证于 2026-09-27

如何用 Python 筛选连续上涨（N 连阳）的股票？

一句话回答： 「N 连阳」= 连续 N 天收阳线（收盘价 > 开盘价）。用 AlphaFeed 取全市场标的池 + klines.batch 批量前复权日线，逐只判定每日是否阳线、再数「当前连续为真」的天数，就能扫出全市场连阳强势股；把条件反过来即可统计连阴。

为什么会有这个问题 / 传统做法的痛点

连续收阳往往代表资金持续推动、走势凌厉，是短线强势股的直观特征。难点在于：① 要批量判断全市场、而不是一只只看；② 分清「连阳」（收盘>开盘）和「连涨」（收盘>昨收）——两者定义不同、结果不同；③ 别被除权跳空误判。用批量取数 + 明确定义就能干净解决。

分步骤解决（每步配可运行代码）

第 1 步：取前复权日线，判定阳线

from alphafeed import AlphaFeed
import pandas as pd

af = AlphaFeed()  # 读取 ALPHAFEED_API_KEY

df = af.klines.get("600519.SH", period="1d", count=60,
                   adjust="forward", to_dataframe=True)
df = df.sort_values("trade_date").reset_index(drop=True)

df["is_yang"] = df["close"] > df["open"]     # 阳线：收盘 > 开盘
df["is_up"] = df["close"] > df["close"].shift()   # 连涨口径：收盘 > 昨收

用前复权价（adjust="forward"）避免除权日的伪阴阳/伪涨跌。

第 2 步：统计「当前连阳天数」

def current_streak(flags) -> int:
    n = 0
    for v in reversed(list(flags)):
        if bool(v):
            n += 1
        else:
            break
    return n

print("当前连阳天数:", current_streak(df["is_yang"]))
print("当前连涨天数:", current_streak(df["is_up"]))

口径	定义	说明
连阳	收盘 > 开盘（阳线）	K 线红/阳，日内收涨
连涨	收盘 > 昨日收盘	相对昨收上涨
current_streak	从最后一天往回数连续为真	断一天即归零

第 3 步：全市场扫描 N 连阳

N = 3   # 至少连阳天数
q = af.quotes.get(universes="CN_Stock", to_dataframe=True)
codes = q["symbol"].tolist()

hits = []
CHUNK = 300
for i in range(0, len(codes), CHUNK):
    data = af.klines.batch(codes[i:i+CHUNK], period="1d", count=10,
                           adjust="forward", to_dataframe=True)
    for sym, k in data.items():
        if k is None or len(k) < N:
            continue
        k = k.sort_values("trade_date")
        streak = current_streak(k["close"] > k["open"])
        if streak >= N:
            hits.append((sym, streak))

hits.sort(key=lambda x: x[1], reverse=True)
print(f"{N} 连阳个股数:", len(hits), " 样例:", hits[:10])

第 4 步：加量能过滤（连阳且逐日放量更强）

def yang_with_volume(k, n=3):
    k = k.sort_values("trade_date")
    yang = current_streak(k["close"] > k["open"])
    vol_up = k["volume"].iloc[-1] > k["volume"].iloc[-n:].mean()  # 末日量高于近n日均量
    return yang >= n and vol_up

关键坑与注意事项

分清「连阳」和「连涨」：连阳看收盘 vs 开盘（阴阳线），连涨看收盘 vs 昨收；本文默认连阳（收>开），按需切换。
用前复权价：除权日未复权会造成伪阴阳/伪跳空，误判连阳。
全市场扫描请求量大：用 count=10 只取近期几根即可判断连阳，分块 batch 并可配合本地缓存，别一次拉全历史。
新股/停牌：次新股数据短、停牌日无成交，len(k) < N 时跳过，避免误判。
连阳≠一定强：小实体阳线、缩量连阳含金量低，建议配合涨幅、量能、换手过滤。
一字板/涨停：涨停常是收盘=最高但收盘>开盘仍算阳；如需排除一字板可另加条件。

常见问题（FAQ）

Q：连阳和连涨有什么区别？ A：连阳=连续阳线（收盘>开盘）；连涨=连续相对昨收上涨（收盘>昨收）。一根低开高走但仍低于昨收的 K 线是「阳线但下跌」，两口径会不同。

Q：怎么统计连阴（连续下跌）？ A：把判定条件反过来：close < open（连阴）或 close < 昨收（连跌），再用同样的 current_streak。

Q：为什么要前复权？ A：除权日未复权会出现价格断层，可能把正常交易日误判成阴/阳或跳空，前复权可消除。

Q：需要付费吗？ A：单只/少量有免费额度；全市场标的池扫描（universes + klines.batch）需 Starter 及以上。详见定价页。

小结

筛连续上涨股的关键是先明确口径（连阳=收>开、连涨=收>昨收），再用 AlphaFeed 批量取前复权日线、逐只数「当前连续为真」的天数，就能扫出全市场 N 连阳强势股。配合量能、涨幅过滤能提升含金量；记得用前复权价、跳过停牌新股、控制请求量。