先搞清楚几个缩写：
TMX - TMX Group Limited
TSX - Toronto Stock Exchange，多伦多证券交易所，股票主板
TSXV - TSX Venture Exchange，多伦多证券交易所创业板
S&P - Standard & Poor's，标准普尔，简称标普
TSX 市场有很多 S&P/TSX 系列指数（S&P/TSX Indices），这些指数列在这个页面的表格中。
表格第一栏是指数信息（Index Information）。
表格第二栏是指数代码和 Quote 页面链接。
表格第三栏是指数编制方法（Methodology）PDF 文件以及 Fact Sheet 的链接（如果有）。
以前点击第一栏里的指数信息，弹出的窗口会列出该指数的全部成分股 （Constituents），还提供 CSV 文件下载。现在有些指数只列出 Top 10 Constituents，或者干脆是空白。
我的目标是取得一个指数的全部成分股数据。
以一个非常重要的指数为例：S&P/TSX 60 Index。
它的 Quote 页面是：https://money.tmx.com/en/quote/%5ETX60
它的成分股页面是：https://money.tmx.com/en/quote/%5ETX60/constituents
这个页面是前端动态渲染的。实际数据来自 TMX 自己的 GraphQL 服务：
https://app-money.tmx.com/graphql。
这个 GraphQL endpoint 目前仍可访问，所以可以直接从 GraphQL
取得完整成分股数据。
我用 Windows 11，通过 Firefox 的 Web Developer Tools 找出 TMX 自己发出的请求。
例如，我当时打开的是：https://money.tmx.com/en/quote/%5ESPTXBMCP/constituents
然后按 F12 → Network，选择 XHR，再按 Ctrl + F5 强制刷新页面，
搜索 graphql。
应该会看到发往
https://app-money.tmx.com/graphql
的 POST 请求。
点开数据量较大的 POST 请求，再看 Request，可以找到类似下面的内容：
{
"operationName": "xxxxx",
"variables": {
...
},
"query": "query xxxxx(...) {...}"
}我截图如下：
点 Response，就可以看到完整的 GraphQL 返回结果，包括全部成分股。
有了 Request 里的数据，就可以完全脱离浏览器，直接用 Python 调用 TMX 的 GraphQL API。
核心接口是：
POST https://app-money.tmx.com/graphql
operationName: getIndexConstituents
variables: {"symbol":"^TX60"}用下面这段代码就可以抓取全部成分股，同时验证实际取得的成分股数量 和 TMX 返回的成分股数量是否一致，并计算权重合计：
import requests
import pandas as pd
URL = "https://app-money.tmx.com/graphql"
payload = {
"operationName": "getIndexConstituents",
"variables": {
"symbol": "^TX60"
},
"query": """
query getIndexConstituents($symbol: String!) {
constituents: getIndexConstituents(symbol: $symbol) {
symbol
quotedMarketValue
longName
shortName
weight
exShortName
exchange
exLongName
__typename
}
keyData: getIndexKeyData(symbol: $symbol) {
adjMarketCap
avgConstituentMarketCap
numConstituents
top10HoldingsAdjMarketCap
ytdPriceReturn
prevDayPriceReturn
prevMonthPriceReturn
prevQuarterPriceReturn
percentWeightLargestConstituent
peRatio
pbRatio
priceToSales
divYield
pcfRatio
__typename
}
}
"""
}
headers = {
"Accept": "*/*",
"Content-Type": "application/json",
"locale": "en",
"Origin": "https://money.tmx.com",
"Referer": "https://money.tmx.com/",
"User-Agent": "Mozilla/5.0"
}
response = requests.post(
URL,
json=payload,
headers=headers,
timeout=30
)
response.raise_for_status()
result = response.json()
# GraphQL 即使返回 HTTP 200，也可能包含 errors
if "errors" in result:
raise RuntimeError(result["errors"])
constituents = result["data"]["constituents"]
key_data = result["data"]["keyData"]
df = pd.DataFrame(constituents)
# 删除 GraphQL 内部字段
df = df.drop(columns=["__typename"], errors="ignore")
# 调整列顺序
df = df[
[
"symbol",
"longName",
"exchange",
"exShortName",
"quotedMarketValue",
"weight",
"shortName",
"exLongName"
]
]
print(df.to_string(index=False))
print("\n------------------------------")
print("TMX reported constituents :", key_data["numConstituents"])
print("Actually downloaded :", len(df))
print("Weight total :", df["weight"].sum())
print("------------------------------")
df.to_csv(
"TX60_constituents.csv",
index=False,
encoding="utf-8-sig"
)
df.to_excel(
"TX60_constituents.xlsx",
index=False
)
print("\nSaved:")
print("TX60_constituents.csv")
print("TX60_constituents.xlsx")再进一步，可以把脚本做成通用版，同时支持命令行参数和无参数时的 交互输入。
既可以这样运行：
python3 tmx_index.py "^SPTXBMCP"也可以直接运行：
python3 tmx_index.py然后根据提示输入指数代码。
也可以不输入 ^：
python3 tmx_index.py TX60脚本会自动补成：
^TX60通用版代码如下：
import argparse
import re
import sys
from datetime import date
from pathlib import Path
import pandas as pd
import requests
# ============================================================
# Configuration
# ============================================================
GRAPHQL_URL = "https://app-money.tmx.com/graphql"
# 权重是百分数，例如 6.717 表示 6.717%
# 因为 TMX 数据有四舍五入，所以允许总权重与 100% 有轻微误差
WEIGHT_TOLERANCE = 0.10
# ============================================================
# GraphQL query
# ============================================================
QUERY = """
query getIndexConstituents($symbol: String!) {
constituents: getIndexConstituents(symbol: $symbol) {
symbol
quotedMarketValue
longName
shortName
weight
exShortName
exchange
exLongName
__typename
}
keyData: getIndexKeyData(symbol: $symbol) {
adjMarketCap
avgConstituentMarketCap
numConstituents
top10HoldingsAdjMarketCap
ytdPriceReturn
prevDayPriceReturn
prevMonthPriceReturn
prevQuarterPriceReturn
percentWeightLargestConstituent
peRatio
pbRatio
priceToSales
divYield
pcfRatio
__typename
}
}
"""
# ============================================================
# Utility functions
# ============================================================
def normalize_symbol(symbol: str) -> str:
"""
清理用户输入的指数代码。
例如：
SPTXBMCP -> ^SPTXBMCP
^SPTXBMCP -> ^SPTXBMCP
"""
symbol = symbol.strip().upper()
if not symbol:
raise ValueError("Index symbol cannot be empty.")
if not symbol.startswith("^"):
symbol = "^" + symbol
return symbol
def symbol_for_filename(symbol: str) -> str:
"""
把指数代码转换成适合文件名的形式。
^SPTXBMCP -> SPTXBMCP
"""
name = symbol.lstrip("^")
# 删除 Windows/macOS 文件名里可能有问题的字符
name = re.sub(r'[<>:"/\\|?*]', "_", name)
return name
# ============================================================
# Download
# ============================================================
def get_index_data(symbol: str) -> dict:
"""
从 TMX GraphQL 下载指数成分股及 key data。
"""
payload = {
"operationName": "getIndexConstituents",
"variables": {
"symbol": symbol
},
"query": QUERY
}
headers = {
"Accept": "*/*",
"Content-Type": "application/json",
"locale": "en",
"Origin": "https://money.tmx.com",
"Referer": "https://money.tmx.com/",
"User-Agent": "Mozilla/5.0"
}
try:
response = requests.post(
GRAPHQL_URL,
json=payload,
headers=headers,
timeout=30
)
response.raise_for_status()
except requests.RequestException as exc:
raise RuntimeError(
f"TMX request failed: {exc}"
) from exc
# 防止 HTTP 200 但实际不是 JSON
try:
result = response.json()
except requests.exceptions.JSONDecodeError as exc:
preview = response.text[:500]
raise RuntimeError(
"TMX returned a non-JSON response.\n\n"
f"Response preview:\n{preview}"
) from exc
# GraphQL 即使 HTTP 200，也可能返回 errors
if result.get("errors"):
raise RuntimeError(
f"TMX GraphQL error:\n{result['errors']}"
)
if "data" not in result:
raise RuntimeError(
"TMX response does not contain a 'data' field."
)
constituents = result["data"].get("constituents")
key_data = result["data"].get("keyData")
if constituents is None:
raise RuntimeError(
f"No constituent data returned for {symbol}."
)
if key_data is None:
raise RuntimeError(
f"No index key data returned for {symbol}."
)
return {
"constituents": constituents,
"keyData": key_data
}
# ============================================================
# Data processing
# ============================================================
def build_constituents_dataframe(constituents: list) -> pd.DataFrame:
"""
将 TMX constituent JSON 转换成 DataFrame。
"""
df = pd.DataFrame(constituents)
if df.empty:
raise RuntimeError("TMX returned an empty constituent list.")
# GraphQL 内部字段，不需要保存
df = df.drop(
columns=["__typename"],
errors="ignore"
)
# 调整列顺序
preferred_columns = [
"symbol",
"longName",
"shortName",
"exShortName",
"exchange",
"exLongName",
"quotedMarketValue",
"weight",
]
available_columns = [
col
for col in preferred_columns
if col in df.columns
]
df = df[available_columns]
return df
# ============================================================
# Validation
# ============================================================
def validate_index(
df: pd.DataFrame,
key_data: dict
) -> dict:
"""
检查：
1. 下载数量是否与 TMX numConstituents 一致
2. 权重是否约等于 100%
3. TSX / TSXV 数量
"""
actual_count = len(df)
reported_count = key_data.get(
"numConstituents"
)
count_match = (
reported_count is not None
and actual_count == reported_count
)
# --------------------------
# Weight
# --------------------------
if "weight" in df.columns:
weight_total = df["weight"].sum()
else:
weight_total = None
if weight_total is not None:
weight_difference = abs(
weight_total - 100.0
)
weight_ok = (
weight_difference
<= WEIGHT_TOLERANCE
)
else:
weight_difference = None
weight_ok = False
# --------------------------
# Exchange counts
# --------------------------
if "exShortName" in df.columns:
exchange_counts = (
df["exShortName"]
.fillna("Unknown")
.value_counts()
.to_dict()
)
else:
exchange_counts = {}
tsx_count = exchange_counts.get(
"TSX",
0
)
tsxv_count = exchange_counts.get(
"TSXV",
0
)
# --------------------------
# Duplicate ticker check
# --------------------------
duplicate_symbols = []
if "symbol" in df.columns:
duplicates = df[
df["symbol"].duplicated(
keep=False
)
]
if not duplicates.empty:
duplicate_symbols = sorted(
duplicates["symbol"]
.dropna()
.unique()
.tolist()
)
return {
"reported_count": reported_count,
"actual_count": actual_count,
"count_match": count_match,
"weight_total": weight_total,
"weight_difference": weight_difference,
"weight_ok": weight_ok,
"tsx_count": tsx_count,
"tsxv_count": tsxv_count,
"exchange_counts": exchange_counts,
"duplicate_symbols": duplicate_symbols,
}
# ============================================================
# Output
# ============================================================
def save_files(
symbol: str,
df: pd.DataFrame,
key_data: dict,
validation: dict
):
"""
输出：
INDEX_constituents_YYYY-MM-DD.csv
INDEX_constituents_YYYY-MM-DD.xlsx
"""
today = date.today().isoformat()
clean_symbol = symbol_for_filename(
symbol
)
base_name = (
f"{clean_symbol}"
f"_constituents_{today}"
)
csv_path = Path(
f"{base_name}.csv"
)
xlsx_path = Path(
f"{base_name}.xlsx"
)
# --------------------------
# CSV
# --------------------------
df.to_csv(
csv_path,
index=False,
encoding="utf-8-sig"
)
# --------------------------
# Excel Summary sheet
# --------------------------
summary_rows = [
["Index Symbol", symbol],
["Download Date", today],
[
"TMX Reported Constituents",
validation["reported_count"]
],
[
"Downloaded Constituents",
validation["actual_count"]
],
[
"Count Match",
validation["count_match"]
],
[
"Weight Total (%)",
validation["weight_total"]
],
[
"Weight Check",
validation["weight_ok"]
],
[
"TSX Constituents",
validation["tsx_count"]
],
[
"TSXV Constituents",
validation["tsxv_count"]
],
]
summary_df = pd.DataFrame(
summary_rows,
columns=[
"Item",
"Value"
]
)
# --------------------------
# Key Data sheet
# --------------------------
clean_key_data = {
k: v
for k, v in key_data.items()
if k != "__typename"
}
key_data_df = pd.DataFrame(
clean_key_data.items(),
columns=[
"Metric",
"Value"
]
)
# --------------------------
# Excel
# --------------------------
with pd.ExcelWriter(
xlsx_path,
engine="openpyxl"
) as writer:
df.to_excel(
writer,
sheet_name="Constituents",
index=False
)
summary_df.to_excel(
writer,
sheet_name="Summary",
index=False
)
key_data_df.to_excel(
writer,
sheet_name="KeyData",
index=False
)
# ----------------------
# 简单格式化
# ----------------------
workbook = writer.book
ws = workbook["Constituents"]
ws.freeze_panes = "A2"
ws.auto_filter.ref = (
ws.dimensions
)
# 调整列宽
widths = {
"A": 14, # symbol
"B": 48, # longName
"C": 24, # shortName
"D": 12, # exShortName
"E": 12, # exchange
"F": 28, # exLongName
"G": 20, # market value
"H": 14, # weight
}
for column, width in widths.items():
ws.column_dimensions[
column
].width = width
# Market Cap number format
if "quotedMarketValue" in df.columns:
market_col = (
df.columns.get_loc(
"quotedMarketValue"
)
+ 1
)
for row in range(
2,
len(df) + 2
):
ws.cell(
row=row,
column=market_col
).number_format = (
'#,##0.00'
)
# Weight format
# 注意 TMX 已经返回 6.717 = 6.717%
# 所以不能用 Excel 百分比格式
if "weight" in df.columns:
weight_col = (
df.columns.get_loc(
"weight"
)
+ 1
)
for row in range(
2,
len(df) + 2
):
ws.cell(
row=row,
column=weight_col
).number_format = (
'0.000'
)
# Summary
ws_summary = workbook["Summary"]
ws_summary.column_dimensions[
"A"
].width = 32
ws_summary.column_dimensions[
"B"
].width = 22
# KeyData
ws_key = workbook["KeyData"]
ws_key.column_dimensions[
"A"
].width = 34
ws_key.column_dimensions[
"B"
].width = 22
return csv_path, xlsx_path
# ============================================================
# Console report
# ============================================================
def print_report(
symbol: str,
validation: dict
):
"""
输出验证结果。
"""
print()
print("=" * 60)
print(f"TMX INDEX: {symbol}")
print("=" * 60)
print(
"TMX reported constituents :",
validation["reported_count"]
)
print(
"Actually downloaded :",
validation["actual_count"]
)
if validation["count_match"]:
print(
"Constituent count check : PASS"
)
else:
print(
"Constituent count check : FAIL"
)
print()
if validation["weight_total"] is not None:
print(
"Weight total : "
f"{validation['weight_total']:.6f}%"
)
if validation["weight_ok"]:
print(
"Weight check : PASS"
)
else:
print(
"Weight check : FAIL"
)
print()
print(
"TSX constituents :",
validation["tsx_count"]
)
print(
"TSXV constituents :",
validation["tsxv_count"]
)
# 其他交易所
other_exchanges = {
exchange: count
for exchange, count
in validation[
"exchange_counts"
].items()
if exchange not in {
"TSX",
"TSXV"
}
}
for exchange, count in (
other_exchanges.items()
):
print(
f"{exchange} constituents"
f"{' ' * max(1, 11 - len(exchange))}:",
count
)
print()
if validation[
"duplicate_symbols"
]:
print(
"WARNING: Duplicate symbols:",
", ".join(
validation[
"duplicate_symbols"
]
)
)
else:
print(
"Duplicate symbol check : PASS"
)
print("=" * 60)
# ============================================================
# Main
# ============================================================
def main():
parser = argparse.ArgumentParser(
description=(
"Download TMX index constituents "
"from TMX Money GraphQL API."
)
)
parser.add_argument(
"symbol",
nargs="?",
help=(
"TMX index symbol, "
"e.g. ^SPTXBMCP"
)
)
args = parser.parse_args()
# 如果命令行没输入，就交互询问
if args.symbol:
input_symbol = args.symbol
else:
input_symbol = input(
"Enter TMX index symbol "
"(e.g. ^SPTXBMCP): "
)
try:
symbol = normalize_symbol(
input_symbol
)
print(
f"\nDownloading {symbol} "
"from TMX..."
)
index_data = get_index_data(
symbol
)
df = (
build_constituents_dataframe(
index_data[
"constituents"
]
)
)
validation = validate_index(
df,
index_data["keyData"]
)
print_report(
symbol,
validation
)
csv_path, xlsx_path = (
save_files(
symbol,
df,
index_data["keyData"],
validation
)
)
print("\nSaved files:")
print(
f" CSV : {csv_path.resolve()}"
)
print(
f" Excel: {xlsx_path.resolve()}"
)
# 如果主要验证失败，给明显提示
if not validation[
"count_match"
]:
print(
"\nWARNING: Downloaded "
"constituent count does "
"not match TMX keyData."
)
if not validation[
"weight_ok"
]:
print(
"\nWARNING: Constituent "
"weights do not add up "
"to approximately 100%."
)
except Exception as exc:
print(
f"\nERROR: {exc}",
file=sys.stderr
)
sys.exit(1)
if __name__ == "__main__":
main()运行时会看到类似下面的结果：
============================================================
TMX INDEX: ^TX60
============================================================
TMX reported constituents : 60
Actually downloaded : 60
Constituent count check : PASS
Weight total : 99.996000%
Weight check : PASS
TSX constituents : 60
TSXV constituents : 0
Duplicate symbol check : PASS
============================================================
Saved files:
CSV : ...\Python_Projects\TMX_Indices\TX60_constituents_2026-08-11.csv
Excel: ...\Python_Projects\TMX_Indices\TX60_constituents_2026-08-11.xlsx生成的 Excel 文件有 3 个 sheet：
Constituents：完整成分股数据
Summary：成分股数量、权重合计以及 TSX/TSXV 数量检查
KeyData：TMX 返回的 P/E、P/B、P/S、Dividend Yield、
YTD Return 等指数数据
No comments:
Post a Comment