A股日线看不出方向？用 Python 多周期 K 线做共振选股，趋势一目了然
只看日线选股有一个致命问题：你不知道自己在做什么级别的交易。
日线金叉了，你买进去，结果周线还在往下走——日线级别的反弹被周线级别的下跌吞掉了。反过来也有：日线死叉了你吓得卖了，结果月线趋势完好，跌下去两天就拉回来。
专业做趋势的人会同时看多个周期。当日线、周线、月线同时指向同一个方向——这叫"多周期共振"——趋势的可靠性会大幅提高。
AlphaFeed 的 K 线接口支持 period="1d" / "1w" / "1M" 三种周期，可以用同一个接口拉日线、周线、月线数据。这篇文章用这三个周期做一套完整的共振选股系统。
1. 什么是多周期共振
简单说：大周期定方向，小周期找时机。
| 周期 | 看什么 | 对应交易级别 |
|---|---|---|
| 月线 | 长期趋势方向 | 中长线（持仓数月） |
| 周线 | 中期趋势强度 | 波段（持仓数周） |
| 日线 | 短期买卖时机 | 短线（持仓数天） |
最理想的买入时机：月线多头 + 周线多头 + 日线刚刚翻多。
2. 单只票的三周期分析
先拿一只票演示，看看三个周期怎么配合：
from alphafeed import AlphaFeed
import pandas as pd
af = AlphaFeed()
symbol = "600519.SH"
# 用同一个接口拉三个周期的数据
df_daily = af.klines.get(symbol, period="1d", count=250, adjust="forward", to_dataframe=True)
df_weekly = af.klines.get(symbol, period="1w", count=100, adjust="forward", to_dataframe=True)
df_monthly = af.klines.get(symbol, period="1M", count=36, adjust="forward", to_dataframe=True)
def analyze_trend(df: pd.DataFrame, label: str) -> dict:
"""分析单个周期的趋势状态"""
df = df.sort_values("trade_date").reset_index(drop=True)
price = df["close"].iloc[-1]
ma5 = df["close"].rolling(5).mean().iloc[-1]
ma10 = df["close"].rolling(10).mean().iloc[-1]
ma20 = df["close"].rolling(min(20, len(df))).mean().iloc[-1]
# 趋势打分
score = 0
details = []
if price > ma5:
score += 1
details.append(f"站上 MA5({ma5:.2f})")
if price > ma10:
score += 1
details.append(f"站上 MA10({ma10:.2f})")
if price > ma20:
score += 1
details.append(f"站上 MA20({ma20:.2f})")
if ma5 > ma10 > ma20:
score += 2
details.append("均线多头排列")
elif ma5 < ma10 < ma20:
score -= 2
details.append("均线空头排列")
return {
"周期": label,
"现价": price,
"得分": score,
"状态": "多头" if score >= 3 else "空头" if score <= -1 else "震荡",
"细节": " | ".join(details),
}
daily_trend = analyze_trend(df_daily, "日线")
weekly_trend = analyze_trend(df_weekly, "周线")
monthly_trend = analyze_trend(df_monthly, "月线")
print(f"=== {symbol} 多周期分析 ===\n")
for t in [monthly_trend, weekly_trend, daily_trend]:
print(f" {t['周期']}: {t['状态']}（得分 {t['得分']}）")
print(f" {t['细节']}")
print()
# 共振判断
states = [monthly_trend["状态"], weekly_trend["状态"], daily_trend["状态"]]
if all(s == "多头" for s in states):
print("🟢 三周期共振多头 —— 趋势非常强")
elif all(s == "空头" for s in states):
print("🔴 三周期共振空头 —— 趋势非常弱")
elif monthly_trend["状态"] == "多头" and weekly_trend["状态"] == "多头":
print("🟡 大周期偏多，日线待确认 —— 可以观察")
else:
print("⚪ 各周期方向不一致 —— 暂时观望")
3. 批量多周期共振选股
对一批股票同时做三周期分析，筛选出"三线共振"的标的：
from alphafeed import AlphaFeed
import pandas as pd
af = AlphaFeed()
# 股票池（可以扩展到几百只）
stock_pool = [
"600519.SH", "000001.SZ", "300750.SZ", "002594.SZ", "601318.SH",
"000858.SZ", "600036.SH", "000333.SZ", "601012.SH", "600276.SH",
"600900.SH", "601398.SH", "600030.SH", "000651.SZ", "002415.SZ",
"600887.SH", "601166.SH", "000568.SZ", "600809.SH", "002304.SZ",
"601888.SH", "600809.SH", "300059.SZ", "002475.SZ", "000725.SZ",
"601899.SH", "600031.SH", "002714.SZ", "600585.SH", "000002.SZ",
]
# 批量拉三个周期的 K 线
print("拉取日线...")
daily_data = af.klines.batch(
stock_pool, period="1d", count=60,
adjust="forward", to_dataframe=True, show_progress=True,
)
print("拉取周线...")
weekly_data = af.klines.batch(
stock_pool, period="1w", count=30,
adjust="forward", to_dataframe=True, show_progress=True,
)
print("拉取月线...")
monthly_data = af.klines.batch(
stock_pool, period="1M", count=12,
adjust="forward", to_dataframe=True, show_progress=True,
)
print(f"\n数据拉取完成: {len(stock_pool)} 只票 × 3 个周期\n")
三个 batch 调用，自动并发，30 只票 × 3 个周期 = 90 次请求，但实际只需要几秒。
4. 定义趋势判断函数
def get_trend_score(df: pd.DataFrame) -> int:
"""
趋势打分: -5 到 +5
正数 = 多头，负数 = 空头，0 附近 = 震荡
"""
if df is None or len(df) < 10:
return 0
df = df.sort_values("trade_date").reset_index(drop=True)
price = df["close"].iloc[-1]
ma5 = df["close"].rolling(5).mean().iloc[-1]
ma10 = df["close"].rolling(10).mean().iloc[-1]
ma20 = df["close"].rolling(min(20, len(df))).mean().iloc[-1]
score = 0
if price > ma5: score += 1
else: score -= 1
if price > ma10: score += 1
else: score -= 1
if price > ma20: score += 1
else: score -= 1
if ma5 > ma10: score += 1
else: score -= 1
if ma10 > ma20: score += 1
else: score -= 1
return score
def classify_trend(score: int) -> str:
if score >= 3:
return "多头"
elif score <= -3:
return "空头"
else:
return "震荡"
5. 筛选共振标的
results = []
for sym in stock_pool:
d_score = get_trend_score(daily_data.get(sym))
w_score = get_trend_score(weekly_data.get(sym))
m_score = get_trend_score(monthly_data.get(sym))
d_trend = classify_trend(d_score)
w_trend = classify_trend(w_score)
m_trend = classify_trend(m_score)
# 综合得分 = 月线权重最大
total_score = m_score * 3 + w_score * 2 + d_score * 1
results.append({
"代码": sym,
"月线": m_trend,
"周线": w_trend,
"日线": d_trend,
"月得分": m_score,
"周得分": w_score,
"日得分": d_score,
"综合分": total_score,
})
rdf = pd.DataFrame(results).sort_values("综合分", ascending=False)
# 三周期共振多头
resonance_bull = rdf[
(rdf["月线"] == "多头") &
(rdf["周线"] == "多头") &
(rdf["日线"] == "多头")
]
# 月线周线多头，日线待确认（潜在买入时机）
potential_buy = rdf[
(rdf["月线"] == "多头") &
(rdf["周线"] == "多头") &
(rdf["日线"] != "多头")
]
# 三周期共振空头
resonance_bear = rdf[
(rdf["月线"] == "空头") &
(rdf["周线"] == "空头") &
(rdf["日线"] == "空头")
]
print(f"=== 多周期共振选股结果 ===\n")
print(f"🟢 三周期共振多头: {len(resonance_bull)} 只")
if len(resonance_bull) > 0:
print(resonance_bull[["代码", "月线", "周线", "日线", "综合分"]].to_string(index=False))
print(f"\n🟡 大周期多头 + 日线待确认: {len(potential_buy)} 只")
if len(potential_buy) > 0:
print(potential_buy[["代码", "月线", "周线", "日线", "综合分"]].to_string(index=False))
print(f"\n🔴 三周期共振空头: {len(resonance_bear)} 只")
if len(resonance_bear) > 0:
print(resonance_bear[["代码", "月线", "周线", "日线", "综合分"]].to_string(index=False))
print(f"\n完整排名:")
print(rdf[["代码", "月线", "周线", "日线", "综合分"]].to_string(index=False))
6. 加入动量确认
光看均线位置还不够，加入动量指标让筛选更精确：
def get_momentum(df: pd.DataFrame, period: int = 20) -> float:
"""计算近 N 根 K 线的动量（涨幅）"""
if df is None or len(df) < period:
return 0
df = df.sort_values("trade_date").reset_index(drop=True)
return df["close"].iloc[-1] / df["close"].iloc[-period] - 1
# 给共振多头的票加上动量排序
if len(resonance_bull) > 0:
momentum_data = []
for _, row in resonance_bull.iterrows():
sym = row["代码"]
mom_d = get_momentum(daily_data.get(sym), 20) # 日线 20 日动量
mom_w = get_momentum(weekly_data.get(sym), 8) # 周线 8 周动量
momentum_data.append({
"代码": sym,
"20日动量": f"{mom_d:+.1%}",
"8周动量": f"{mom_w:+.1%}",
"综合分": row["综合分"],
})
mom_df = pd.DataFrame(momentum_data)
print(f"\n🟢 共振多头标的动量排序:")
print(mom_df.to_string(index=False))
7. 一个更实际的选股流程
把多周期共振嵌入日常选股工作流：
# resonance_scan.py
"""多周期共振选股扫描"""
from alphafeed import AlphaFeed
import pandas as pd
af = AlphaFeed()
def resonance_scan(symbols: list) -> pd.DataFrame:
"""对一组标的做多周期共振扫描"""
daily = af.klines.batch(
symbols, period="1d", count=60,
adjust="forward", to_dataframe=True,
)
weekly = af.klines.batch(
symbols, period="1w", count=30,
adjust="forward", to_dataframe=True,
)
monthly = af.klines.batch(
symbols, period="1M", count=12,
adjust="forward", to_dataframe=True,
)
results = []
for sym in symbols:
ds = get_trend_score(daily.get(sym))
ws = get_trend_score(weekly.get(sym))
ms = get_trend_score(monthly.get(sym))
dt = classify_trend(ds)
wt = classify_trend(ws)
mt = classify_trend(ms)
resonance = "共振多头" if dt == wt == mt == "多头" else \
"共振空头" if dt == wt == mt == "空头" else \
"大周期多头" if mt == "多头" and wt == "多头" else \
"方向不一致"
results.append({
"代码": sym,
"月线": mt,
"周线": wt,
"日线": dt,
"共振": resonance,
"综合分": ms * 3 + ws * 2 + ds,
})
return pd.DataFrame(results).sort_values("综合分", ascending=False)
# 使用
my_stocks = ["600519.SH", "000001.SZ", "300750.SZ", "002594.SZ",
"601318.SH", "000858.SZ", "600036.SH", "000333.SZ"]
result = resonance_scan(my_stocks)
print(result.to_string(index=False))
8. 为什么这个方法有效
多周期共振不是万能的，但它解决了一个很实际的问题：降低逆势交易的概率。
| 场景 | 只看日线 | 三周期共振 |
|---|---|---|
| 日线金叉买入，周线还在跌 | 会买 ❌ | 不买 ✅ |
| 日线死叉卖出，月线趋势完好 | 会卖 ❌ | 不卖 ✅ |
| 三个周期都翻多 | 可能买也可能没注意 | 明确信号 ✅ |
| 月线空头反弹 | 可能当反转 ❌ | 大周期空头，不参与 ✅ |
本质上，它帮你过滤掉了大量"日线看着不错但大周期不配合"的假信号。
9. AlphaFeed 在这件事里的价值
多周期分析需要同时拉日线、周线、月线三套数据。用 AlphaFeed 做这件事很顺畅：
# 同一个接口，只改 period 参数
df_d = af.klines.get(sym, period="1d", count=60, adjust="forward", to_dataframe=True)
df_w = af.klines.get(sym, period="1w", count=30, adjust="forward", to_dataframe=True)
df_m = af.klines.get(sym, period="1M", count=12, adjust="forward", to_dataframe=True)
不需要自己从日线合成周线月线（很多数据源只提供日线，周线月线要自己算——日期对齐、跨周处理、节假日，坑很多）。AlphaFeed 直接给你计算好的周线和月线 K 线，开高低收都是准的。
再加上 batch 批量接口，30 只票 × 3 个周期 = 90 次请求，SDK 内部并发处理，几秒钟全部拉完。
- AlphaFeed 官网：https://alphafeed.org/
- Python SDK 快速开始：https://docs.alphafeed.org/zh-Hans/sdk/python-quickstart