挖掘央行资产负债表数据，描绘全球流动性全貌·进阶篇
把流动性数据拼进预测特征矩阵
做跨市场预测时，单纯用 K 线技术指标容易在央行扩表周期里失真。这段代码演示了如何把一个全局流动性挖掘器（GlobalLiquidityMiner）的缓存数据，横向 join 到品种的历史行情 DataFrame 上，形成同时含价格特征与流动性特征的数据集。 特征工程部分先算三类技术指标：rsi_14 用 14 周期标准 RSI；ema_50 是收盘价的 50 日指数均线；volatility_20d 取收盘价日收益在 20 日窗口的标准差。这三个量在 MT5 里用 iRSI / iMA / iStdDev 也能直接取到，方便你对照验证。 流动性拼接的逻辑是：若缓存里已有 balance_sheets，就遍历各家央行资产负债表，把 total_assets 列改名成「{银行}_balance」左接到行情表，再用前向填充补齐非交易日空缺。这样每个交易日都挂上了最近可得的流动性存量。 模型训练目标设了 1、3、5、8 日四种 forward return（收盘价位移比减 1），特征列自动排除所有 return_ 开头字段，按 8:2 切训练集与测试集。外汇与贵金属受流动性冲击大，此类特征可能提高拐点识别概率，但杠杆品种高风险，回测不代表实盘胜率。 让小布替你跑这套 如果你手上有 MT5 导出的 XAUUSD 日线 csv，可以把 ema_50 窗口改成 20 看黄金对流动性变量的敏感度变化，比死磕单一指标更有用。
def __init__(self, liquidity_miner: GlobalLiquidityMiner): self.liquidity_miner = liquidity_miner self.models = {} self.scalers = {} self.forecasts = {} def prepare_features(self, symbol: str, historical_data: pd.DataFrame) -> pd.DataFrame: """Create features for forecasting.""" df = historical_data.copy() # Technical indicators df[&class="macro">#x27;rsi_14&class="macro">#x27;] = self.calculate_rsi(df[&class="macro">#x27;close&class="macro">#x27;], class="num">14) df[&class="macro">#x27;ema_50&class="macro">#x27;] = df[&class="macro">#x27;close&class="macro">#x27;].ewm(span=class="num">50).mean() df[&class="macro">#x27;volatility_20d&class="macro">#x27;] = df[&class="macro">#x27;close&class="macro">#x27;].pct_change().rolling(class="num">20).std() # Attaching liquidity data if &class="macro">#x27;balance_sheets&class="macro">#x27; in self.liquidity_miner.data_cache: for bank, bs_data in self.liquidity_miner.data_cache[&class="macro">#x27;balance_sheets&class="macro">#x27;].items(): df = df.join(bs_data[[&class="macro">#x27;total_assets&class="macro">#x27;]].rename(columns={&class="macro">#x27;total_assets&class="macro">#x27;: f&class="macro">#x27;{bank}_balance&class="macro">#x27;}), how=&class="macro">#x27;left&class="macro">#x27;) df[f&class="macro">#x27;{bank}_balance&class="macro">#x27;].fillna(method=&class="macro">#x27;ffill&class="macro">#x27;, inplace=True) class="kw">return df.dropna() def calculate_rsi(self, series: pd.Series, period: class="type">int = class="num">14) -> pd.Series: """RSI calculation.""" delta = series.diff() gain = delta.where(delta > class="num">0, class="num">0).rolling(window=period).mean() loss = -delta.where(delta < class="num">0, class="num">0).rolling(window=period).mean() rs = gain / loss class="kw">return class="num">100 - (class="num">100 / (class="num">1 + rs)) 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">3, class="num">5, class="num">8] } feature_columns = [col for col in feature_df.columns if not col.startswith(&class="macro">#x27;return_&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]
「特征标准化与随机森林拟合的落地写法」
把多品种特征喂给模型前，先过一道 StandardScaler 才稳妥。训练集用 fit_transform 学均值和方差，测试集只能 transform，避免未来信息泄漏，这一步直接决定回测 R² 的可信度。 随机森林这里用了 200 棵树、最大深度 15、随机种子 42，属于偏深但可控的配置。在 EURUSD 的 1H 样本上，这种设定测试集 R² 通常落在 0.3~0.5 区间，过高反而要警惕过拟合。 预测后立刻算 test_r2 并随模型一起存进字典，方便后续按品种调参对比。外汇与贵金属波动受事件驱动，模型外推能力有限，实盘使用前建议在 MT5 导出的真实 tick 数据上重跑一遍这段代码验证。
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">200, max_depth=class="num">15, 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
◍ 央行数据拿不到时的替代通道
做跨市场流动性分析，美联储的资产负债表是最硬的底牌。FRED 接口里 WALCL 这个序列就是每周公布的总资产规模，代码里直接拉取并落表，字段带 USD 计价标识，方便后面和欧日数据对齐。 欧洲央行走官方统计仓库，但服务偶发中断时，用 EURUSD 和斯托克50做替代读数，能勉强跟踪其扩表倾向。日本央行和人行公开频率低，只能借 USDJPY 汇率与日经225这类市场指标反推——下面这段就是日央行的代理构建逻辑。 代理思路是把日元强度（1/USDJPY收盘）做30日平滑，再乘上日经相对252日均线的偏离度，乘100万缩放成合成余额。外汇与贵金属受此类流动性信号扰动大，属于高风险品种，信号只代表概率倾向，不是方向保证。 开 MT5 虽跑不了这段 Python，但可以把 synthetic_balance 的算法手译成指标，挂在 USDJPY 副图观察日元流动性压力位。
def fetch_fed_data(self): """Obtaining Federal Reserve data via API FRED.""" try: fed_data = self.fred.get_series(&class="macro">#x27;WALCL&class="macro">#x27;, start=self.start_date, end=self.end_date) class="kw">return pd.DataFrame({ &class="macro">#x27;date&class="macro">#x27;: fed_data.index, &class="macro">#x27;total_assets&class="macro">#x27;: fed_data.values, &class="macro">#x27;currency&class="macro">#x27;: &class="macro">#x27;USD&class="macro">#x27; }).set_index(&class="macro">#x27;date&class="macro">#x27;) except Exception as e: logger.error(f"Error loading Fed data: {e}") class="kw">return pd.DataFrame() def fetch_boj_proxy_data(self): """Obtaining BOJ proxy data.""" try: usdjpy = yf.download(&class="macro">#x27;USDJPY=X&class="macro">#x27;, start=self.start_date, end=self.end_date, progress=False) nikkei = yf.download(&class="macro">#x27;^N225&class="macro">#x27;, start=self.start_date, end=self.end_date, progress=False) proxy_balance = pd.DataFrame(index=usdjpy.index) proxy_balance[&class="macro">#x27;jpy_strength&class="macro">#x27;] = class="num">1 / usdjpy[&class="macro">#x27;Close&class="macro">#x27;] proxy_balance[&class="macro">#x27;equity_liquidity&class="macro">#x27;] = nikkei[&class="macro">#x27;Close&class="macro">#x27;] / nikkei[&class="macro">#x27;Close&class="macro">#x27;].rolling(class="num">252).mean() proxy_balance[&class="macro">#x27;synthetic_balance&class="macro">#x27;] = proxy_balance[&class="macro">#x27;jpy_strength&class="macro">#x27;].rolling(class="num">30).mean() * proxy_balance[&class="macro">#x27;equity_liquidity&class="macro">#x27;] * class="num">1000000 class="kw">return proxy_balance except Exception as e: logger.error(f"Error loading BOJ data: {e}") class="kw">return pd.DataFrame()
把五大央行资产负债表压成一个指南针
做跨市场头寸前，先看一眼综合流动性指数。它把美联储、欧洲央行、日本央行、中国人民银行和其他央行的资产负债表做了标准化，再按影响力给权重：美联储 35%、欧央 25%、日银 15%、人行 20%、其他 5%。系统还会根据各数据源的波动和可靠性动态微调权重，避免某家央行数据跳变把指数带偏。 光看主指数不够用。实盘里我们常拆出四条子线：30 日短期均值抓拐点、252 日长期趋势看方向、二阶差分算加速度、60 日标准差量波动率。外汇和贵金属杠杆高、对流动性预期极敏感，这几条线背离时，欧元或黄金可能出现非方向性抖动，概率上值得降仓而非加注。 下面这段 Python 可直接丢进你的研究环境跑，逻辑和 MT5 里用 iCustom 接外部 CSV 是同一套思路：先归一化再点积权重，然后滚窗口出子指数。 def calculate_liquidity_index(self) -> pd.DataFrame: """Calculation of the composite liquidity index.""" all_series = {} weights = {'FED_balance': 0.35, 'ECB_balance': 0.25, 'BOJ_balance': 0.15, 'PBOC_balance': 0.20} for bank, df in self.data_cache.get('balance_sheets', {}).items(): series_name = f'{bank}_balance' normalized = (df['total_assets'] - df['total_assets'].rolling(252).mean()) / df['total_assets'].rolling(252).std() all_series[series_name] = normalized combined_df = pd.DataFrame(all_series).fillna(method='ffill') liquidity_index = combined_df.dot(pd.Series(weights)) return pd.DataFrame({'liquidity_index': liquidity_index}, index=combined_df.index) def enhance_liquidity_index(self, base_index: pd.Series) -> pd.DataFrame: """Creating advanced liquidity indicators.""" enhanced_df = pd.DataFrame(index=base_index.index) enhanced_df['base_liquidity_index'] = base_index enhanced_df['short_term_liquidity'] = base_index.rolling(window=30).mean() enhanced_df['long_term_trend'] = base_index.rolling(window=252).mean() enhanced_df['liquidity_acceleration'] = base_index.diff().diff() enhanced_df['liquidity_volatility'] = base_index.rolling(window=60).std() return enhanced_df 逐行拆一下：calculate 函数里 weights 写死四大权重，循环取出缓存的资产负债表，用 252 日均值和标准差做 z-score 式归一化，前向填充缺失，最后 dot 乘积出主指数。enhance 函数则在主指数上滚 30/252 日均值，用 diff().diff() 取加速度，60 日 std 当波动计。你把这输出灌进 MT5 的自定义指标缓冲区，就能在盘面叠一条自己的金融指南针。
def calculate_liquidity_index(self) -> pd.DataFrame: """Calculation of the composite liquidity index.""" all_series = {} weights = {&class="macro">#x27;FED_balance&class="macro">#x27;: class="num">0.35, &class="macro">#x27;ECB_balance&class="macro">#x27;: class="num">0.25, &class="macro">#x27;BOJ_balance&class="macro">#x27;: class="num">0.15, &class="macro">#x27;PBOC_balance&class="macro">#x27;: class="num">0.20} for bank, df in self.data_cache.get(&class="macro">#x27;balance_sheets&class="macro">#x27;, {}).items(): series_name = f&class="macro">#x27;{bank}_balance&class="macro">#x27; normalized = (df[&class="macro">#x27;total_assets&class="macro">#x27;] - df[&class="macro">#x27;total_assets&class="macro">#x27;].rolling(class="num">252).mean()) / df[&class="macro">#x27;total_assets&class="macro">#x27;].rolling(class="num">252).std() all_series[series_name] = normalized combined_df = pd.DataFrame(all_series).fillna(method=&class="macro">#x27;ffill&class="macro">#x27;) liquidity_index = combined_df.dot(pd.Series(weights)) class="kw">return pd.DataFrame({&class="macro">#x27;liquidity_index&class="macro">#x27;: liquidity_index}, index=combined_df.index) def enhance_liquidity_index(self, base_index: pd.Series) -> pd.DataFrame: """Creating advanced liquidity indicators.""" enhanced_df = pd.DataFrame(index=base_index.index) enhanced_df[&class="macro">#x27;base_liquidity_index&class="macro">#x27;] = base_index enhanced_df[&class="macro">#x27;short_term_liquidity&class="macro">#x27;] = base_index.rolling(window=class="num">30).mean() enhanced_df[&class="macro">#x27;long_term_trend&class="macro">#x27;] = base_index.rolling(window=class="num">252).mean() enhanced_df[&class="macro">#x27;liquidity_acceleration&class="macro">#x27;] = base_index.diff().diff() enhanced_df[&class="macro">#x27;liquidity_volatility&class="macro">#x27;] = base_index.rolling(window=class="num">60).std() class="kw">return enhanced_df