挖掘央行资产负债表数据，描绘全球流动性全貌（基础篇）
「从央行资产负债表看全球流动性水泵」
2008年次贷危机与2020年后的市场动荡，让央行工具箱从单纯调利率扩到企业债购买、货币互换、为财政融资等非常规操作。这些动作像巨型水泵，向经济体注水或抽水，效果直接写在央行资产负债表的规模与结构上——它是货币政策最诚实的晴雨表。 传统技术派看K线猜方向，如同凭云层断天气；纯基本面又太慢，跟不上短线开仓节奏。把美联储、ECB、BOJ、PBoC四家表内数据汇总成全球流动性综合指标，才有机会打通两层壁垒，用流动性读数去压住短期波动与长期趋势的判断。 这套思路在MT5里不是空谈：用MQL5拉取宏观数据做特征，再接机器学习模块，能捕捉传统指标漏掉的隐藏节律。外汇与贵金属受此驱动明显，但杠杆品种高风险，流动性拐点只代表概率倾斜，不等于方向锁死。
◍ 流动性怎么牵动汇率与趋势
全球流动性不是简单的货币总量，而是货币、金融工具和资本流转机制叠在一起的体系。狭义看它是资产无损变现的能力，放到全球尺度就是资本跨国跨市场流动的难易。资本一旦持续挪窝，长期趋势往往就被催生出来。 美联储、欧央、日银、人行是四足调控主体。看央行资产负债表别只当财报，它直接显示购债、放贷投了多少钱进去：美联储买国债扩表就是放流动性，欧央提准备金要求就是在收。 扩表通常压本币汇率，逻辑就两条——供给多了按供需压价值，且宽松常配降息削弱资产吸引力。但这不是死规矩：主要央行同步扩表时，多币同贬，货币对波动会明显收窄。 真正要看的是相对扩表速度。当日银扩表增速远超美联储时，日元相对美元走弱的概率偏大。传导渠道有四个：利率（降息利空汇率）、资产配置（购债逼资金转场）、信贷（环境好改跨境流）、预期（表态提前引导）。 预期管理已是明牌工具。美联储主席暗示收紧，市场可能在决议前就抛美元。所以盯流动性，资产负债表数值和央行措辞得两头看，后者影响力不亚于实际落地。外汇与贵金属受此驱动波动剧烈，属高风险博弈，验证前先用 MT5 周线把四大央行资产负债表变化叠加上去比一比。
把央行资产负债表喂给模型的采集层
做宏观流动性驱动的外汇策略，第一步不是跑模型，而是把各国央行资产负债表弄成能对齐的时间序列。美联储 FRED 有周度报告（如 WALCL 总资产），中国人民银行是季度披露，日本央行格式复杂，三者的发布频率和计价货币都不同，直接拼会错位。 GlobalLiquidityMiner 这块做的就是脏活：拉数、插值补缺失、标准化、按时间对齐。它对接 FRED API，也能处理 yfinance 和 requests 抓的非标源，把不同频率的数据重采样到统一时间轴，给后面的预测模块当底料。 代码里 FED 分支拉了 WALCL（总资产）、WSHOSHO（持仓证券）、WLRRAL（贷款与回购），并用 pct_change(periods=52) 算同比52周增速——这是把周度序列转成年度视图，避免短期噪声盖掉流动性趋势。 模块解耦是重点：采集层和 ForexLiquidityForecaster 预测层独立，你换了日央行数据源只改 Miner，不碰随机森林那层。外汇和贵金属受宏观流动性影响大但高杠杆高风险，这类信号只提高概率倾向，不能当方向保证。 让小布替你跑这套：开 MT5 或本地 Python，先只接 FRED 的 WALCL，看 assets_growth_rate 转负时 EURUSD 随后一个月走弱的概率，再决定要不要加央行维度。
class="kw">import pandas as pd class="kw">import logging from typing class="kw">import Dict from fredapi class="kw">import Fred class="kw">import yfinance as yf class="kw">import requests from io class="kw">import StringIO logger = logging.getLogger(__name__) class GlobalLiquidityMiner: def __init__(self, fred_api_key: str, start_date: str, end_date: str): self.fred = Fred(api_key=fred_api_key) if fred_api_key else None self.start_date = start_date self.end_date = end_date self.data_cache = {} def fetch_central_bank_balance_sheets(self) -> Dict[str, pd.DataFrame]: """Obtaining central bank balance sheet data.""" balance_sheets = {} if self.fred: logger.info("Loading Federal Reserve data...") try: fed_total_assets = self.fred.get_series(&class="macro">#x27;WALCL&class="macro">#x27;, start=self.start_date, end=self.end_date) fed_securities = self.fred.get_series(&class="macro">#x27;WSHOSHO&class="macro">#x27;, start=self.start_date, end=self.end_date) fed_loans = self.fred.get_series(&class="macro">#x27;WLRRAL&class="macro">#x27;, start=self.start_date, end=self.end_date) balance_sheets[&class="macro">#x27;FED&class="macro">#x27;] = pd.DataFrame({ &class="macro">#x27;date&class="macro">#x27;: fed_total_assets.index, &class="macro">#x27;total_assets&class="macro">#x27;: fed_total_assets.values, &class="macro">#x27;securities_held&class="macro">#x27;: fed_securities.reindex(fed_total_assets.index, method=&class="macro">#x27;ffill&class="macro">#x27;).values, &class="macro">#x27;loans_and_repos&class="macro">#x27;: fed_loans.reindex(fed_total_assets.index, method=&class="macro">#x27;ffill&class="macro">#x27;).values, &class="macro">#x27;currency&class="macro">#x27;: &class="macro">#x27;USD&class="macro">#x27; }) balance_sheets[&class="macro">#x27;FED&class="macro">#x27;][&class="macro">#x27;assets_growth_rate&class="macro">#x27;] = balance_sheets[&class="macro">#x27;FED&class="macro">#x27;][&class="macro">#x27;total_assets&class="macro">#x27;].pct_change(periods=class="num">52) balance_sheets[&class="macro">#x27;FED&class="macro">#x27;][&class="macro">#x27;securities_share&class="macro">#x27;] = balance_sheets[&class="macro">#x27;FED&class="macro">#x27;][&class="macro">#x27;securities_held&class="macro">#x27;] / balance_sheets[&class="macro">#x27;FED&class="macro">#x27;][&class="macro">#x27;total_assets&class="macro">#x27;] logger.info(f"Loaded {len(fed_total_assets)} Federal Reserve data entries")
「用随机森林给流动性建模」
把全球流动性数据接进来之后，下一步是训练一个能预测汇率回报的模型。下面这段 Python 类直接以 1 日、5 日收盘回报率为目标变量，用随机森林回归去拟合特征矩阵。 特征列会先剔除所有以 return_、volatility_、direction_ 开头的字段，避免标签泄漏。训练集按 80% 切分，剩余 20% 用于测 R²，RandomForestRegressor 固定 n_estimators=100、max_depth=10、random_state=42，保证结果可复现。 每个标的（symbol）的预测模型与对应的 StandardScaler 都存进实例字典，后续换周期或换品种时直接调 self.models[symbol] 即可。外汇与贵金属受流动性驱动明显，但模型输出只是概率倾向，实盘仍属高风险，建议先拿历史数据跑通再上 MT5 验证。
from sklearn.ensemble class="kw">import RandomForestRegressor from sklearn.preprocessing class="kw">import StandardScaler from sklearn.metrics class="kw">import r2_score, mean_squared_error class="kw">import numpy as np class ForexLiquidityForecaster: def __init__(self, liquidity_miner: GlobalLiquidityMiner): self.liquidity_miner = liquidity_miner self.models = {} self.scalers = {} def build_prediction_model(self, symbol: str, feature_df: pd.DataFrame): """ Training a forecasting model.""" targets = { f&class="macro">#x27;return_{h}d&class="macro">#x27;: feature_df[&class="macro">#x27;close&class="macro">#x27;].shift(-h) / feature_df[&class="macro">#x27;close&class="macro">#x27;] - class="num">1 for h in [class="num">1, class="num">5] } feature_columns = [col for col in feature_df.columns if not col.startswith((&class="macro">#x27;return_&class="macro">#x27;, &class="macro">#x27;volatility_&class="macro">#x27;, &class="macro">#x27;direction_&class="macro">#x27;))] X = feature_df[feature_columns].dropna() train_size = class="type">int(len(X) * class="num">0.8) X_train, X_test = X.iloc[:train_size], X.iloc[train_size:] models = {} for target_name, target_series in targets.items(): y = target_series.dropna() common_idx = X.index.intersection(y.index) X_aligned, y_aligned = X.loc[common_idx], y.loc[common_idx] scaler = StandardScaler() X_train_scaled = scaler.fit_transform(X_aligned.iloc[:train_size]) X_test_scaled = scaler.transform(X_aligned.iloc[train_size:]) model = RandomForestRegressor(n_estimators=class="num">100, max_depth=class="num">10, random_state=class="num">42) model.fit(X_train_scaled, y_aligned.iloc[:train_size]) test_pred = model.predict(X_test_scaled) test_r2 = r2_score(y_aligned.iloc[train_size:], test_pred) models[target_name] = {&class="macro">#x27;model&class="macro">#x27;: model, &class="macro">#x27;scaler&class="macro">#x27;: scaler, &class="macro">#x27;r2&class="macro">#x27;: test_r2} self.models[symbol] = models self.scalers[symbol] = scaler