威廉·江恩(William Gann)方法(第三部分):占星术是否有效?·综合运用
(3/3)· 当古老星象遇上机器学习,EURUSD走势里究竟藏着几分天体信号
◍ 用星体数据喂CatBoost试水价格预测
想把天文周期和外盘报价扯上关系,最直接的就是把行星坐标、月相、太阳活动当成特征丢进机器学习模型。这里跑了两套:一套CatBoost回归盯收盘价,一套分类判方向,数据源混了MT5导出的金融序列和skyfield算出的星历。 结果很骨感——分类模型准确率只略高于50%,基本和抛硬币一个水平;回归那边也没认真调参,R²低到没引用价值。外汇和贵金属本身高波动、受宏观突发驱动,单靠星体lag特征想稳定捕捉价格偏向,概率上目前看不成立。 代码里值得抄的是特征构造套路:金融列做1~5期lag,七颗行星的ra/dec也做同结构lag,再用LabelEncoder把行星相位字符串转数值,缺失值填列均值后dropna。回归模型用test_size=0.3、random_state=1切分,iterations=500、depth=9,early_stopping_rounds=200,跑完直接打印MSE/MAE/R²和特征重要性。开MT5导出CSV,换自己的品种复一遍,能立刻看到哪颗星在模型里权重虚高。
class="kw">import pandas as pd class="kw">import numpy as np from catboost class="kw">import CatBoostRegressor from sklearn.model_selection class="kw">import train_test_split from sklearn.metrics class="kw">import mean_squared_error, mean_absolute_error, r2_score class="kw">import matplotlib.pyplot as plt from sklearn.preprocessing class="kw">import LabelEncoder # Loading data data = pd.read_csv(&class="macro">#x27;merged_astro_financial_data.csv&class="macro">#x27;) # Converting date to class="type">class="kw">datetime data[&class="macro">#x27;date&class="macro">#x27;] = pd.to_datetime(data[&class="macro">#x27;date&class="macro">#x27;]) # Creating lags for financial data for col in [&class="macro">#x27;open&class="macro">#x27;, &class="macro">#x27;high&class="macro">#x27;, &class="macro">#x27;low&class="macro">#x27;, &class="macro">#x27;close&class="macro">#x27;]: for lag in range(class="num">1, class="num">6): # Creating lags from class="num">1 to class="num">5 data[f&class="macro">#x27;{col}_lag{lag}&class="macro">#x27;] = data[col].shift(lag) # Creating lags for astronomical data astro_cols = [&class="macro">#x27;mercury&class="macro">#x27;, &class="macro">#x27;venus&class="macro">#x27;, &class="macro">#x27;mars&class="macro">#x27;, &class="macro">#x27;jupiter&class="macro">#x27;, &class="macro">#x27;saturn&class="macro">#x27;, &class="macro">#x27;uranus&class="macro">#x27;, &class="macro">#x27;neptune&class="macro">#x27;] for col in astro_cols: data[f&class="macro">#x27;{col}_ra&class="macro">#x27;] = data[col].apply(lambda x: eval(x)[&class="macro">#x27;ra&class="macro">#x27;] if pd.notna(x) else np.nan) data[f&class="macro">#x27;{col}_dec&class="macro">#x27;] = data[col].apply(lambda x: eval(x)[&class="macro">#x27;dec&class="macro">#x27;] if pd.notna(x) else np.nan) for lag in range(class="num">1, class="num">6): # Lags from class="num">1 to class="num">5 data[f&class="macro">#x27;{col}_ra_lag{lag}&class="macro">#x27;] = data[f&class="macro">#x27;{col}_ra&class="macro">#x27;].shift(lag) data[f&class="macro">#x27;{col}_dec_lag{lag}&class="macro">#x27;] = data[f&class="macro">#x27;{col}_dec&class="macro">#x27;].shift(lag) data.drop(columns=[col, f&class="macro">#x27;{col}_ra&class="macro">#x27;, f&class="macro">#x27;{col}_dec&class="macro">#x27;], inplace=True) # Converting aspects to numerical features aspect_cols = [&class="macro">#x27;mercury_saturn&class="macro">#x27;, &class="macro">#x27;venus_mars&class="macro">#x27;, &class="macro">#x27;venus_jupiter&class="macro">#x27;, &class="macro">#x27;venus_uranus&class="macro">#x27;, &class="macro">#x27;mars_jupiter&class="macro">#x27;, &class="macro">#x27;mars_uranus&class="macro">#x27;, &class="macro">#x27;jupiter_uranus&class="macro">#x27;, &class="macro">#x27;mercury_neptune&class="macro">#x27;, &class="macro">#x27;venus_saturn&class="macro">#x27;, &class="macro">#x27;venus_neptune&class="macro">#x27;, &class="macro">#x27;mars_saturn&class="macro">#x27;, &class="macro">#x27;mercury_venus&class="macro">#x27;, &class="macro">#x27;mars_neptune&class="macro">#x27;, &class="macro">#x27;mercury_uranus&class="macro">#x27;, &class="macro">#x27;saturn_neptune&class="macro">#x27;, &class="macro">#x27;mercury_jupiter&class="macro">#x27;, &class="macro">#x27;mercury_mars&class="macro">#x27;, &class="macro">#x27;jupiter_saturn&class="macro">#x27;] # Using LabelEncoder for encoding aspects label_encoders = {} for col in aspect_cols: label_encoders[col] = LabelEncoder() data[col] = label_encoders[col].fit_transform(data[col].astype(str)) # Filling missing values with mean values for numeric columns numeric_cols = data.select_dtypes(include=[np.number]).columns data[numeric_cols] = data[numeric_cols].fillna(data[numeric_cols].mean()) # Removing rows with missing values data = data.dropna() # Preparing features and target variable features = [col for col in data.columns if col not in [&class="macro">#x27;date&class="macro">#x27;, &class="macro">#x27;time&class="macro">#x27;, &class="macro">#x27;close&class="macro">#x27;]] X = data[features] y = data[&class="macro">#x27;close&class="macro">#x27;] # Splitting data into training and testing sets X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=class="num">0.3, random_state=class="num">1) # Creating and training the CatBoost model model = CatBoostRegressor(iterations=class="num">500, learning_rate=class="num">0.1, depth=class="num">9, random_state=class="num">1) model.fit(X_train, y_train, eval_set=(X_test, y_test), early_stopping_rounds=class="num">200, verbose=class="num">100) # Evaluating the model y_pred = model.predict(X_test) mse = mean_squared_error(y_test, y_pred) mae = mean_absolute_error(y_test, y_pred) r2 = r2_score(y_test, y_pred) print(f"Mean Squared Error: {mse}") print(f"Mean Absolute Error: {mae}") print(f"R-squared Score: {r2}") # Visualizing feature importance feature_importance = model.feature_importances_ feature_names = X.columns sorted_idx = np.argsort(feature_importance) pos = np.arange(sorted_idx.shape[class="num">0]) + class="num">0.5 plt.figure(figsize=(class="num">12, class="num">6)) plt.barh(pos, feature_importance[sorted_idx], align=&class="macro">#x27;center&class="macro">#x27;) plt.yticks(pos, np.array(feature_names)[sorted_idx]) plt.xlabel(&class="macro">#x27;Feature Importance&class="macro">#x27;) plt.title(&class="macro">#x27;Feature Importance&class="macro">#x27;) plt.show() # Predicting the next value def predict_next(): # Selecting the last row of data last_data = data.iloc[-class="num">1] input_features = last_data[features].values.reshape(class="num">1, -class="num">1) # Prediction prediction = model.predict(input_features) print(f"Prediction for the next closing price: {prediction[class="num">0]}") # Example of class="kw">using the function to predict the next value predict_next() class="kw">import pandas as pd class="kw">import numpy as np from skyfield.api class="kw">import load, utc from class="type">class="kw">datetime class="kw">import class="type">class="kw">datetime, timedelta class="kw">import requests class="kw">import MetaTrader5 as mt5 class="kw">import seaborn as sns class="kw">import matplotlib.pyplot as plt from catboost class="kw">import CatBoostClassifier from sklearn.model_selection class="kw">import train_test_split from sklearn.metrics class="kw">import accuracy_score, classification_report, confusion_matrix from sklearn.preprocessing class="kw">import LabelEncoder # Part class="num">1: Collecting astronomical data planets = load(&class="macro">#x27;de421.bsp&class="macro">#x27;) earth = planets[&class="macro">#x27;earth&class="macro">#x27;] ts = load.timescale() def get_planet_positions(date): t = ts.from_datetime(date.replace(tzinfo=utc)) planet_positions = {} planet_ids = { &class="macro">#x27;mercury&class="macro">#x27;: &class="macro">#x27;MERCURY BARYCENTER&class="macro">#x27;, &class="macro">#x27;venus&class="macro">#x27;: &class="macro">#x27;VENUS BARYCENTER&class="macro">#x27;, &class="macro">#x27;mars&class="macro">#x27;: &class="macro">#x27;MARS BARYCENTER&class="macro">#x27;, &class="macro">#x27;jupiter&class="macro">#x27;: &class="macro">#x27;JUPITER BARYCENTER&class="macro">#x27;, &class="macro">#x27;saturn&class="macro">#x27;: &class="macro">#x27;SATURN BARYCENTER&class="macro">#x27;, &class="macro">#x27;uranus&class="macro">#x27;: &class="macro">#x27;URANUS BARYCENTER&class="macro">#x27;, &class="macro">#x27;neptune&class="macro">#x27;: &class="macro">#x27;NEPTUNE BARYCENTER&class="macro">#x27; } for planet, planet_id in planet_ids.items(): planet_obj = planets[planet_id] astrometric = earth.at(t).observe(planet_obj) ra, dec, _ = astrometric.radec() planet_positions[planet] = {&class="macro">#x27;ra&class="macro">#x27;: ra.hours, &class="macro">#x27;dec&class="macro">#x27;: dec.degrees} class="kw">return planet_positions def get_moon_phase(date): t = ts.from_datetime(date.replace(tzinfo=utc)) eph = load(&class="macro">#x27;de421.bsp&class="macro">#x27;) moon, sun, earth = eph[&class="macro">#x27;moon&class="macro">#x27;], eph[&class="macro">#x27;sun&class="macro">#x27;], eph[&class="macro">#x27;earth&class="macro">#x27;] e = earth.at(t) _, m, _ = e.observe(moon).apparent().ecliptic_latlon() _, s, _ = e.observe(sun).apparent().ecliptic_latlon() phase = (m.degrees - s.degrees) % class="num">360 class="kw">return phase def get_solar_activity(date): url = f"https:class=class="str">"cmt">//services.swpc.noaa.gov/json/solar-cycle/observed-solar-cycle-indices.json" response = requests.get(url) data = response.json() target_date = date.strftime("%Y-%m") closest_data = min(data, key=lambda x: abs(class="type">class="kw">datetime.strptime(x[&class="macro">#x27;time-tag&class="macro">#x27;], "%Y-%m") - class="type">class="kw">datetime.strptime(target_date, "%Y-%m"))) class="kw">return { &class="macro">#x27;sunspot_number&class="macro">#x27;: closest_data.get(&class="macro">#x27;ssn&class="macro">#x27;, None), &class="macro">#x27;f10.7_flux&class="macro">#x27;: closest_data.get(&class="macro">#x27;f10.class="num">7&class="macro">#x27;, None) } def calculate_aspects(positions): aspects = {} planets = list(positions.keys()) for i in range(len(planets)): for j in range(i+class="num">1, len(planets)): planet1 = planets[i] planet2 = planets[j] ra1 = positions[planet1][&class="macro">#x27;ra&class="macro">#x27;] ra2 = positions[planet2][&class="macro">#x27;ra&class="macro">#x27;] angle = abs(ra1 - ra2) % class="num">24 angle = min(angle, class="num">24 - angle) * class="num">15 # Convert to degrees if abs(angle - class="num">0) <= class="num">10 or abs(angle - class="num">180) <= class="num">10:
把星历与 EURUSD 日线喂给 CatBoost 跑分类
这段脚本把前面算好的天文特征(行星赤经赤纬、相位、太阳活动)和 MT5 拉取的 EURUSD 日线对齐,再做滞后特征后训练二分类模型,判断次日收盘价相对当日是涨是跌。外汇与贵金属属高风险品种,以下结论仅描述代码逻辑与回测框架,不预示任何收益。 数据同步用 inner join 把 financial_df 的 time 与 astro_df 的 date 对齐,缺失交易日直接丢弃;天文列里的字典先拆出 ra/dec 再建 1~5 日滞后,金融的 open/high/low/close 同样做 5 阶 lag,数值空值用列均值填。 目标变量写成 price_change = (close.shift(-1) > close).astype(int),即次日收盘高于当日记为 1。训练集测试集按 7:3 切分,random_state 固定为 1;CatBoost 设 iterations=500、learning_rate=0.1、depth=9,early_stopping_rounds=200。 跑完会打印 accuracy、classification_report 与 confusion_matrix,并画 feature_importance 水平条形图。样本区间写死为 2023-03-01 至 2024-07-30,符号 EURUSD、周期 TIMEFRAME_D1,你改 start_date 或 symbol 就能在 MT5 环境里复跑验证哪类天文特征权重高。
aspects[f"{planet1}_{planet2}"] = "conjunction" if abs(angle - class="num">0) <= class="num">10 else "opposition"
elif abs(angle - class="num">90) <= class="num">10:
aspects[f"{planet1}_{planet2}"] = "square"
elif abs(angle - class="num">120) <= class="num">10:
aspects[f"{planet1}_{planet2}"] = "trine"
class="kw">return aspects
# Part class="num">2: Obtaining financial data through MetaTrader5
def get_financial_data(symbol, start_date, end_date):
if not mt5.initialize():
print("initialize() failed")
mt5.shutdown()
class="kw">return None
timeframe = mt5.TIMEFRAME_D1
rates = mt5.copy_rates_range(symbol, timeframe, start_date, end_date)
mt5.shutdown()
financial_df = pd.DataFrame(rates)
financial_df[&class="macro">#x27;time&class="macro">#x27;] = pd.to_datetime(financial_df[&class="macro">#x27;time&class="macro">#x27;], unit=&class="macro">#x27;s&class="macro">#x27;)
class="kw">return financial_df
# Part class="num">3: Synchronizing astronomical and financial data
def sync_data(astro_df, financial_df):
astro_df[&class="macro">#x27;date&class="macro">#x27;] = pd.to_datetime(astro_df[&class="macro">#x27;date&class="macro">#x27;]).dt.tz_localize(None)
financial_df[&class="macro">#x27;time&class="macro">#x27;] = pd.to_datetime(financial_df[&class="macro">#x27;time&class="macro">#x27;])
merged_data = pd.merge(financial_df, astro_df, left_on=&class="macro">#x27;time&class="macro">#x27;, right_on=&class="macro">#x27;date&class="macro">#x27;, how=&class="macro">#x27;inner&class="macro">#x27;)
class="kw">return merged_data
# Part class="num">4: Training the model and making predictions
def train_and_predict(merged_data):
# Converting aspects to numerical features
aspect_cols = [col for col in merged_data.columns if &class="macro">#x27;_&class="macro">#x27; in col and col not in [&class="macro">#x27;date&class="macro">#x27;, &class="macro">#x27;time&class="macro">#x27;]]
label_encoders = {}
for col in aspect_cols:
label_encoders[col] = LabelEncoder()
merged_data[col] = label_encoders[col].fit_transform(merged_data[col].astype(str))
# Creating lags for financial data
for col in [&class="macro">#x27;open&class="macro">#x27;, &class="macro">#x27;high&class="macro">#x27;, &class="macro">#x27;low&class="macro">#x27;, &class="macro">#x27;close&class="macro">#x27;]:
for lag in range(class="num">1, class="num">6):
merged_data[f&class="macro">#x27;{col}_lag{lag}&class="macro">#x27;] = merged_data[col].shift(lag)
# Creating lags for astronomical data
astro_cols = [&class="macro">#x27;mercury&class="macro">#x27;, &class="macro">#x27;venus&class="macro">#x27;, &class="macro">#x27;mars&class="macro">#x27;, &class="macro">#x27;jupiter&class="macro">#x27;, &class="macro">#x27;saturn&class="macro">#x27;, &class="macro">#x27;uranus&class="macro">#x27;, &class="macro">#x27;neptune&class="macro">#x27;]
for col in astro_cols:
merged_data[f&class="macro">#x27;{col}_ra&class="macro">#x27;] = merged_data[col].apply(lambda x: eval(x)[&class="macro">#x27;ra&class="macro">#x27;] if pd.notna(x) else np.nan)
merged_data[f&class="macro">#x27;{col}_dec&class="macro">#x27;] = merged_data[col].apply(lambda x: eval(x)[&class="macro">#x27;dec&class="macro">#x27;] if pd.notna(x) else np.nan)
for lag in range(class="num">1, class="num">6):
merged_data[f&class="macro">#x27;{col}_ra_lag{lag}&class="macro">#x27;] = merged_data[f&class="macro">#x27;{col}_ra&class="macro">#x27;].shift(lag)
merged_data[f&class="macro">#x27;{col}_dec_lag{lag}&class="macro">#x27;] = merged_data[f&class="macro">#x27;{col}_dec&class="macro">#x27;].shift(lag)
merged_data.drop(columns=[col, f&class="macro">#x27;{col}_ra&class="macro">#x27;, f&class="macro">#x27;{col}_dec&class="macro">#x27;], inplace=True)
# Filling missing values with mean values for numeric columns
numeric_cols = merged_data.select_dtypes(include=[np.number]).columns
merged_data[numeric_cols] = merged_data[numeric_cols].fillna(merged_data[numeric_cols].mean())
merged_data = merged_data.dropna()
# Creating binary target variable
merged_data[&class="macro">#x27;price_change&class="macro">#x27;] = (merged_data[&class="macro">#x27;close&class="macro">#x27;].shift(-class="num">1) > merged_data[&class="macro">#x27;close&class="macro">#x27;]).astype(class="type">int)
# Removing rows with missing values in the target variable
merged_data = merged_data.dropna(subset=[&class="macro">#x27;price_change&class="macro">#x27;])
features = [col for col in merged_data.columns if col not in [&class="macro">#x27;date&class="macro">#x27;, &class="macro">#x27;time&class="macro">#x27;, &class="macro">#x27;close&class="macro">#x27;, &class="macro">#x27;price_change&class="macro">#x27;]]
X = merged_data[features]
y = merged_data[&class="macro">#x27;price_change&class="macro">#x27;]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=class="num">0.3, random_state=class="num">1)
model = CatBoostClassifier(iterations=class="num">500, learning_rate=class="num">0.1, depth=class="num">9, random_state=class="num">1)
model.fit(X_train, y_train, eval_set=(X_test, y_test), early_stopping_rounds=class="num">200, verbose=class="num">100)
y_pred = model.predict(X_test)
accuracy = accuracy_score(y_test, y_pred)
clf_report = classification_report(y_test, y_pred)
conf_matrix = confusion_matrix(y_test, y_pred)
print(f"Accuracy: {accuracy}")
print("Classification Report:")
print(clf_report)
print("Confusion Matrix:")
print(conf_matrix)
# Visualizing feature importance
feature_importance = model.feature_importances_
feature_names = X.columns
sorted_idx = np.argsort(feature_importance)
pos = np.arange(sorted_idx.shape[class="num">0]) + class="num">0.5
plt.figure(figsize=(class="num">12, class="num">6))
plt.barh(pos, feature_importance[sorted_idx], align=&class="macro">#x27;center&class="macro">#x27;)
plt.yticks(pos, np.array(feature_names)[sorted_idx])
plt.xlabel(&class="macro">#x27;Feature Importance&class="macro">#x27;)
plt.title(&class="macro">#x27;Feature Importance&class="macro">#x27;)
plt.show()
# Predicting the next value
def predict_next():
last_data = merged_data.iloc[-class="num">1]
input_features = last_data[features].values.reshape(class="num">1, -class="num">1)
prediction = model.predict(input_features)
print(f"Price change prediction(class="num">0: will decrease, class="num">1: will increase): {prediction[class="num">0]}")
predict_next()
# Main program
start_date = class="type">class="kw">datetime(class="num">2023, class="num">3, class="num">1)
end_date = class="type">class="kw">datetime(class="num">2024, class="num">7, class="num">30)
astro_data = []
current_date = start_date
while current_date <= end_date:
planet_positions = get_planet_positions(current_date)
moon_phase = get_moon_phase(current_date)
try:
solar_activity = get_solar_activity(current_date)
except Exception as e:
print(f"Error getting solar activity for {current_date}: {e}")
solar_activity = {&class="macro">#x27;sunspot_number&class="macro">#x27;: None, &class="macro">#x27;f10.7_flux&class="macro">#x27;: None}
aspects = calculate_aspects(planet_positions)
astro_data.append({
&class="macro">#x27;date&class="macro">#x27;: current_date,
&class="macro">#x27;mercury&class="macro">#x27;: str(planet_positions[&class="macro">#x27;mercury&class="macro">#x27;]),
&class="macro">#x27;venus&class="macro">#x27;: str(planet_positions[&class="macro">#x27;venus&class="macro">#x27;]),
&class="macro">#x27;mars&class="macro">#x27;: str(planet_positions[&class="macro">#x27;mars&class="macro">#x27;]),
&class="macro">#x27;jupiter&class="macro">#x27;: str(planet_positions[&class="macro">#x27;jupiter&class="macro">#x27;]),
&class="macro">#x27;saturn&class="macro">#x27;: str(planet_positions[&class="macro">#x27;saturn&class="macro">#x27;]),
&class="macro">#x27;uranus&class="macro">#x27;: str(planet_positions[&class="macro">#x27;uranus&class="macro">#x27;]),
&class="macro">#x27;neptune&class="macro">#x27;: str(planet_positions[&class="macro">#x27;neptune&class="macro">#x27;]),
&class="macro">#x27;moon_phase&class="macro">#x27;: moon_phase,
**solar_activity,
**aspects
})
current_date += timedelta(days=class="num">1)
astro_df = pd.DataFrame(astro_data)
symbol = "EURUSD"
financial_data = get_financial_data(symbol, start_date, end_date)
if financial_data is not None:
merged_data = sync_data(astro_df, financial_data)
train_and_predict(merged_data)「占星因子在EURUSD上的实测落点」
把天文坐标塞进两套 CatBoost 模型后,EURUSD 的收盘价预测没给出任何值得交易的信号。相关性矩阵里,行星位置与日收盘价的系数全部低于 0.3,按经验阈值这连弱相关都够不上,基本可以判定恒星排布和汇价走势没有可量化纽带。 回归任务的最终 MSE、MAE 与 R² 都弱到难以解释数据方差,特征重要性排序里领跑的是价格滞后项与前期收盘价,而不是任何行星经度。换句话说,比起盯着太阳系轨道,价格自身的历史惯性反而是更靠谱的输入。 分类模型做涨跌二分类,准确率仅勉强越过 50% 这道随机线。对外汇及贵金属这类高杠杆品种而言,这种表现意味着占星类特征不具备边际信息,实盘若据此开仓只会放大高风险下的随机亏损。
◍ 把工具请下神坛
回测与相关性检验给出的信号很直接:用天文数据套进实际行情去做预测,统计上基本不成立。作者原话是占星类方法「完全无效」,连江恩那套被包装成秘籍的路线,也更像卖课话术而非可复现的边缘。 但别把江恩角度一棍打死。同一份研究里能看到,价格对角度变化确有反应,只是单独拿来当预测器不够。把它降维成特征丢进模型训练,反而可能是条还没走完的路——作者说会自己建数据集接着试。 外汇和贵金属本身就是高杠杆、高波动品种,任何「角度圣杯」若没过样本外验证,直接上实盘就是给券商送点差。工具该留在工作台,不该供上神坛。