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

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

然后你发现，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. 跨市场取数据：同一个函数，三个市场

```python
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`。你不需要做字段映射。

批量获取也一样：

```python
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. 同一个策略，三个市场跑一遍

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

```python
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))
```

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

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

## 4. 按市场分组分析

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

```python
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 只，等权配置：

```python
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 股龙头和它们的海外对标公司的实时表现：

```python
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 拼起来，日期会错位。处理方法：

```python
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 股有效的行业轮动信号，在美股是否也有效？ |

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

## 结语

跨市场量化的门槛过去很高，主要高在"数据统一"上。光是把 A 股和美股的 K 线数据格式统一成相同的字段名、时间格式、复权方式，就够折腾一阵子。

AlphaFeed 把这一步省掉了。不管你传 `600519.SH` 还是 `AAPL.US` 还是 `00700.HK`，返回的 DataFrame 结构完全一样。这意味着你的策略函数只需要写一次，就可以在三个市场上跑。

这篇文章展示的是基础框架。在此基础上，你可以做更复杂的事情：跨市场因子分析、全球配置回测、市场间传导关系研究。但所有这些的起点都一样——先把数据拿到手，而且格式要一致。


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

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

编程语言

算法

大数据

"Python跨市场量化实战：一套代码搞定A股、美股、港股数据获取与策略回测。AlphaFeed统一数据接口，无需适配代码，支持600519.SH、AAPL.US等全市场标的。涵盖动量策略实现、多市场表现对比、组合配置等核心场景，解决交易日历、货币单位等跨市场难题，助力量化研究效率提升。"

数据管道

Python

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

一套代码跑遍 A 股、美股、港股：Python 跨市场量化实战-腾讯云开发者社区-腾讯云

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

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

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐