一套代码跑遍 A 股、美股、港股：Python 跨市场量化实战

2601_96288572

218人浏览 · 2026-06-13 20:47:06

2601_96288572 · 2026-06-13 20:47:06 发布

做量化的人迟早会遇到一个问题：我想看看这个策略在美股上表现怎么样。

然后你发现，A 股数据用的是 akshare，美股数据要换成 yfinance，港股又得去找别的接口。三个数据源的代码格式、字段名、时区处理全不一样，光是把三份数据统一成同一个 DataFrame 结构就已经消耗掉半天。

更烦的是，当你终于把数据拼好，发现策略代码里到处都是 if market == "us" 的条件分支。维护两周后代码已经不可读了。

AlphaFeed 的设计目标之一就是解决这个问题：不管是 A 股、美股还是港股，SDK 接口完全一样，返回的字段名完全一样。你不需要写一行跨市场适配代码。

这篇文章从实际需求出发，展示如何用一套代码完成跨市场的数据获取、策略回测和对比分析。

1. 统一的标的代码格式

AlphaFeed 对所有市场使用统一的 代码.交易所 格式：

市场	格式	示例
A 股（上交所）	`CODE.SH`	`600519.SH`（贵州茅台）
A 股（深交所）	`CODE.SZ`	`000858.SZ`（五粮液）
A 股（北交所）	`CODE.BJ`	`830799.BJ`
美股	`TICKER.US`	`AAPL.US`、`TSLA.US`、`MSFT.US`
港股	`CODE.HK`	`00700.HK`（腾讯）、`09988.HK`（阿里巴巴）

这意味着你的策略代码里不需要写 if "SH" in symbol: use_cn_api() 这种逻辑。同一个 af.klines.get() 调用，传不同的 symbol 就行。

2. 跨市场取数据：同一个函数，三个市场

from alphafeed import AlphaFeed

af = AlphaFeed()

symbols = {
    "A股_茅台": "600519.SH",
    "美股_苹果": "AAPL.US",
    "港股_腾讯": "00700.HK",
}

for label, sym in symbols.items():
    df = af.klines.get(
        sym,
        period="1d",
        count=250,
        adjust="forward",
        to_dataframe=True,
    )
    df = df.sort_values("trade_date").reset_index(drop=True)
    print(f"\n{label} ({sym})")
    print(f"  数据量: {len(df)} 条")
    print(f"  起始日: {df['trade_date'].iloc[0]}")
    print(f"  最新日: {df['trade_date'].iloc[-1]}")
    print(f"  最新收盘价: {df['close'].iloc[-1]}")

返回的 DataFrame 字段完全一致：trade_date、open、high、low、close、volume、amount。你不需要做字段映射。

批量获取也一样：

all_symbols = ["600519.SH", "000858.SZ", "AAPL.US", "MSFT.US", "00700.HK", "09988.HK"]

dfs = af.klines.batch(
    all_symbols,
    period="1d",
    count=250,
    adjust="forward",
    to_dataframe=True,
    show_progress=True,
)

for sym, df in dfs.items():
    print(f"{sym}: {len(df)} 条")

3. 同一个策略，三个市场跑一遍

我们写一个简单的动量策略，然后在三个市场的标的上分别跑，看效果差异：

import pandas as pd
import numpy as np
from alphafeed import AlphaFeed

af = AlphaFeed()

def momentum_strategy(df: pd.DataFrame, lookback: int = 20, hold_days: int = 5) -> dict:
    """
    动量策略：如果过去 lookback 天收益为正，持有 hold_days 天。
    """
    df = df.sort_values("trade_date").reset_index(drop=True).copy()
    df["ret"] = df["close"].pct_change()
    df["momentum"] = df["close"].pct_change(lookback)

    df["signal"] = (df["momentum"] > 0).astype(int)
    df["position"] = df["signal"].shift(1).fillna(0)

    df["strategy_ret"] = df["position"] * df["ret"]
    df["equity"] = (1 + df["strategy_ret"]).cumprod()
    df["buyhold"] = (1 + df["ret"].fillna(0)).cumprod()

    total_ret = df["equity"].iloc[-1] - 1
    bh_ret = df["buyhold"].iloc[-1] - 1
    peak = df["equity"].cummax()
    max_dd = (df["equity"] / peak - 1).min()
    sharpe = (
        df["strategy_ret"].mean() / df["strategy_ret"].std() * np.sqrt(252)
        if df["strategy_ret"].std() > 0
        else 0
    )

    return {
        "total_return": round(total_ret, 4),
        "buyhold_return": round(bh_ret, 4),
        "max_drawdown": round(max_dd, 4),
        "sharpe": round(sharpe, 2),
        "trade_days": len(df),
    }


test_symbols = {
    "贵州茅台": "600519.SH",
    "中国平安": "601318.SH",
    "比亚迪": "002594.SZ",
    "苹果": "AAPL.US",
    "微软": "MSFT.US",
    "特斯拉": "TSLA.US",
    "腾讯": "00700.HK",
    "美团": "03690.HK",
    "小米": "01810.HK",
}

all_syms = list(test_symbols.values())

klines = af.klines.batch(
    all_syms,
    period="1d",
    count=500,
    adjust="forward",
    to_dataframe=True,
    show_progress=True,
)

results = []
for name, sym in test_symbols.items():
    if sym not in klines:
        continue
    metrics = momentum_strategy(klines[sym])
    metrics["name"] = name
    metrics["symbol"] = sym
    metrics["market"] = sym.split(".")[-1]
    results.append(metrics)

result_df = pd.DataFrame(results)
print(result_df[["name", "symbol", "market", "total_return", "buyhold_return", "max_drawdown", "sharpe"]].to_string(index=False))

运行结果会告诉你几件事：

这个动量策略在不同市场上的表现差异有多大
策略是否跑赢了买入持有（如果连买入持有都跑不过，策略价值存疑）
不同市场的回撤特征是否不同

4. 按市场分组分析

跑完之后，我们可以按市场维度做聚合分析：

summary = result_df.groupby("market").agg(
    avg_return=("total_return", "mean"),
    avg_sharpe=("sharpe", "mean"),
    avg_drawdown=("max_drawdown", "mean"),
    count=("symbol", "count"),
).round(4)

print("\n=== 按市场汇总 ===")
print(summary)

这张表可以帮你判断：同一个策略逻辑，是 A 股更适合、美股更适合、还是港股更适合？

一个常见的发现是：简单的动量策略在美股大盘股上往往比 A 股表现更稳，因为美股的趋势延续性更强、散户占比更低。但这不是绝对的——不同时间段、不同行情环境下结论可能相反。

5. 跨市场组合：分散配置

如果你有多市场访问权限，一个自然的延伸是做跨市场组合——不把所有鸡蛋放在一个市场的篮子里。

最简单的方法：从每个市场选动量最强的 N 只，等权配置：

import pandas as pd
import numpy as np
from alphafeed import AlphaFeed

af = AlphaFeed()

portfolio_symbols = {
    "CN": ["600519.SH", "000858.SZ", "601318.SH", "002594.SZ", "000001.SZ"],
    "US": ["AAPL.US", "MSFT.US", "GOOGL.US", "AMZN.US", "TSLA.US"],
    "HK": ["00700.HK", "09988.HK", "03690.HK", "01810.HK", "02318.HK"],
}

all_syms = [s for group in portfolio_symbols.values() for s in group]

klines = af.klines.batch(
    all_syms,
    period="1d",
    count=250,
    adjust="forward",
    to_dataframe=True,
    show_progress=True,
)

returns_dict = {}
for sym, df in klines.items():
    df = df.sort_values("trade_date").reset_index(drop=True)
    df["ret"] = df["close"].pct_change()
    returns_dict[sym] = df.set_index("trade_date")["ret"]

returns_df = pd.DataFrame(returns_dict).dropna()

n_assets = len(returns_df.columns)
equal_weight = 1.0 / n_assets

portfolio_ret = returns_df.mean(axis=1)
portfolio_equity = (1 + portfolio_ret).cumprod()

total_return = portfolio_equity.iloc[-1] - 1
max_dd = (portfolio_equity / portfolio_equity.cummax() - 1).min()
sharpe = portfolio_ret.mean() / portfolio_ret.std() * np.sqrt(252) if portfolio_ret.std() > 0 else 0

print(f"跨市场等权组合")
print(f"  标的数量: {n_assets}")
print(f"  累计收益: {total_return:.2%}")
print(f"  最大回撤: {max_dd:.2%}")
print(f"  年化夏普: {sharpe:.2f}")

for market, syms in portfolio_symbols.items():
    cols = [s for s in syms if s in returns_df.columns]
    if not cols:
        continue
    market_ret = returns_df[cols].mean(axis=1)
    market_equity = (1 + market_ret).cumprod()
    print(f"\n  {market} 子组合:")
    print(f"    累计收益: {market_equity.iloc[-1] - 1:.2%}")
    print(f"    最大回撤: {(market_equity / market_equity.cummax() - 1).min():.2%}")

跨市场组合的核心逻辑是相关性——如果 A 股和美股的走势不完全同步，组合在一起能降低整体波动。

6. 跨市场实时行情对比

除了历史回测，盘中观察也是跨市场研究的重要场景。

比如，你想同时看 A 股龙头和它们的海外对标公司的实时表现：

from alphafeed import AlphaFeed

af = AlphaFeed()

pairs = [
    ("比亚迪 A", "002594.SZ", "特斯拉", "TSLA.US"),
    ("腾讯", "00700.HK", "Meta", "META.US"),
    ("阿里巴巴 HK", "09988.HK", "亚马逊", "AMZN.US"),
    ("宁德时代", "300750.SZ", "松下", "PCRFY.US"),
]

all_symbols = []
for pair in pairs:
    all_symbols.extend([pair[1], pair[3]])

quotes = af.quotes.get(symbols=all_symbols, to_dataframe=True)

for cn_name, cn_sym, us_name, us_sym in pairs:
    cn_row = quotes[quotes["symbol"] == cn_sym]
    us_row = quotes[quotes["symbol"] == us_sym]

    cn_chg = (
        (cn_row["last_price"].values[0] - cn_row["prev_close"].values[0])
        / cn_row["prev_close"].values[0]
        if not cn_row.empty
        else None
    )
    us_chg = (
        (us_row["last_price"].values[0] - us_row["prev_close"].values[0])
        / us_row["prev_close"].values[0]
        if not us_row.empty
        else None
    )

    cn_str = f"{cn_chg:+.2%}" if cn_chg is not None else "N/A"
    us_str = f"{us_chg:+.2%}" if us_chg is not None else "N/A"

    print(f"{cn_name}({cn_sym}) {cn_str}  vs  {us_name}({us_sym}) {us_str}")

这种对标分析在以下场景很有用：

美股隔夜大涨/大跌后，判断 A 股/港股对标标的的开盘方向
同行业公司在不同市场的估值分歧
跨市场套利信号的初步观察

7. 跨市场的几个注意事项

交易日历不同

A 股和美股的交易日不一样（节假日、休市安排不同）。直接把两个市场的日 K 拼起来，日期会错位。处理方法：

cn_df = klines["600519.SH"].set_index("trade_date")["close"].rename("maotai")
us_df = klines["AAPL.US"].set_index("trade_date")["close"].rename("aapl")

merged = pd.concat([cn_df, us_df], axis=1)

merged = merged.ffill().dropna()
print(f"交集交易日: {len(merged)} 天")

ffill() 是最简单的处理方式：如果某天一个市场休市，用它最近一个交易日的收盘价填充。对于日线级别的分析，这个误差可以接受。

货币单位不同

A 股价格是人民币，美股是美元，港股是港币。如果你要计算跨市场组合的总收益，需要考虑汇率。但对于单标的策略对比（只看涨跌幅百分比和夏普比率），可以不做货币换算。

盘中时段不同

A 股 9:30–15:00（UTC+8），美股 9:30–16:00（EST），港股 9:30–16:00（UTC+8）。如果你做实时监控，需要注意当前哪个市场在交易时段。AlphaFeed 的实时行情会正确返回最新可用价格，非交易时段返回的是上一个交易日的收盘数据。

8. 用跨市场数据做的几个研究方向

方向	思路
跨市场动量溢出	美股昨天涨的行业，A 股今天相关板块是否跟涨？
波动率比较	同一只公司的 A 股和港股（如 AH 股），波动率差异多大？
市场间相关性变化	A 股和美股的相关性在牛市/熊市中是否不同？
全球配置回测	60% A 股 + 30% 美股 + 10% 港股的组合，年化风险收益比如何？
行业轮动跨市场验证	一个在 A 股有效的行业轮动信号，在美股是否也有效？

这些方向的共同前提是：你能用一套代码快速拿到多个市场的数据，而不是每换一个市场就重写一遍数据管道。