🔭

威廉·江恩（William Gann）方法（第三部分）：占星术是否有效？·进阶篇

（2/3）· 跳过玄学争论，直接拉取天文与EURUSD数据做统计交叉，看行星相位是否真有信号

实战向第 2/3 篇

不少交易者把江恩占星当神秘圣杯，凭几张星图就重仓押注外汇波段。但没做数据同步和显著性检验前，那些'巧合'大概率只是认知偏差。本篇接上篇，用代码把天文与行情拉到同一时间轴，先验证再谈信仰。

本章目录

星体相位与天文数据的批量抓取逻辑
用 Python 直连 MT5 拉 EURUSD 日线
把星历表和 EURUSD 拼到同一张表
用相关系数扫一遍天文与欧元报价
把天文滞后项塞进相关性热力图

◍ 星体相位与天文数据的批量抓取逻辑

这段脚本把星体夹角映射成交易者可用的相位标签：0° 与 180° 附近 ±10° 判定为合相或冲相，90° ±10° 为刑相，120° ±10° 为三合相。容差锁死在 10° 意味着，若两星体夹角落在 10.1° 就不会被计入合相，实际回测时这个边界会直接砍掉一批弱信号。主循环从 2024-04-01 跑到 2024-05-31，按天步进调用星体位置、月相与太阳活动接口；太阳活动取数失败时用 None 兜底，避免整段中断。每天拼成一条记录塞进列表，最后用 pandas 转 DataFrame 并写出 CSV。注意文件名叫 astronomical_data_2018_2024.csv，但循环区间只有 2024 年两个月，直接信文件名会误以为有六年数据。外汇与贵金属受天文周期扰动的概率较高但绝不 deterministic，拿去跑 MT5 前先确认数据窗口和你的样本期对齐。

MQL5 / C++

    if abs(angle - class="num">0) <= class="num">10 or abs(angle - class="num">180) <= class="num">10:
        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
start_date = class="type">class="kw">datetime(class="num">2024, class="num">4, class="num">1, tzinfo=utc)
end_date = class="type">class="kw">datetime(class="num">2024, class="num">5, class="num">31, tzinfo=utc)
current_date = start_date
astronomical_data = []
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)
    
    data = {
        &class="macro">#x27;date&class="macro">#x27;: current_date,
        &class="macro">#x27;moon_phase&class="macro">#x27;: moon_phase,
        &class="macro">#x27;sunspot_number&class="macro">#x27;: solar_activity.get(&class="macro">#x27;sunspot_number&class="macro">#x27;),
        &class="macro">#x27;f10.7_flux&class="macro">#x27;: solar_activity.get(&class="macro">#x27;f10.7_flux&class="macro">#x27;),
        **planet_positions,
        **aspects
    }
    astronomical_data.append(data)
    
    current_date += timedelta(days=class="num">1)
    print(f"Processed: {current_date}")
# Convert data to DataFrame
df = pd.DataFrame(astronomical_data)
# Save data to CSV file
df.to_csv(&class="macro">#x27;astronomical_data_2018_2024.csv&class="macro">#x27;, index=False)
print("Data saved to astronomical_data_2018_2024.csv")

用 Python 直连 MT5 拉 EURUSD 日线

想把经纪商服务器的历史报价抓到本地做回测或特征工程，最直接的是用 Python 的 MetaTrader 5 官方库，它绕开中间商直接读终端缓存。下面这段脚本拉的是 EURUSD 的 D1 数据，时间窗从 2018-01-01 到 2024-12-31，总共约 1800+ 根日 K，足以覆盖近七年主要宏观波动。脚本先 mt5.initialize() 建连，失败就打印并 shutdown 退出，避免挂死进程；copy_rates_range 按品种+周期+起止日抓 OHLCV 数组，再塞进 pandas 的 DataFrame，把 time 字段由 Unix 秒转成可读 datetime，最后落盘成 EURUSD_data.csv，index 不写冗余行。实盘前注意：外汇和贵金属属高杠杆品种，历史数据频率、点差和隔夜息在极端行情会断层，本地 CSV 不等于实时盘口，验证策略请先在 MT5 策略测试器用相同品种周期跑一遍。

MQL5 / C++

class="kw">import MetaTrader5 as mt5
class="kw">import pandas as pd
from class="type">class="kw">datetime class="kw">import class="type">class="kw">datetime
# Connect to MetaTrader5
if not mt5.initialize():
    print("initialize() failed")
    mt5.shutdown()
# Set query parameters
symbol = "EURUSD"
timeframe = mt5.TIMEFRAME_D1
start_date = class="type">class="kw">datetime(class="num">2018, class="num">1, class="num">1)
end_date = class="type">class="kw">datetime(class="num">2024, class="num">12, class="num">31)
# Request historical data
rates = mt5.copy_rates_range(symbol, timeframe, start_date, end_date)
# Convert data to DataFrame
df = pd.DataFrame(rates)
df[&class="macro">#x27;time&class="macro">#x27;] = pd.to_datetime(df[&class="macro">#x27;time&class="macro">#x27;], unit=&class="macro">#x27;s&class="macro">#x27;)
# Save data to CSV file
df.to_csv(f&class="macro">#x27;{symbol}_data.csv&class="macro">#x27;, index=False)
# Terminate the connection to MetaTrader5
mt5.shutdown()

「把星历表和 EURUSD 拼到同一张表」

天文数据有了，EURUSD 的报价序列也有了，下一步是让两套数据按交易日对齐。只有落在同一根 K 线上的行星位置才有意义，否则你拿 2019 年的水星逆行去解释 2023 年的欧元波动，纯属噪声。脚本先分别读入 astronomical_data_2018_2024.csv 和 EURUSD_data.csv，把各自的日期列统一转成 datetime 类型，再用 inner join 按时间对齐。只保留两边都有记录的日期，避免空值污染后续分析。合并结果写回 merged_astro_financial_data.csv，不带索引。你在 MT5 里导出的 EURUSD 若是 M1 或 H1 周期，注意天文数据若是日频，join 后同一天会复制给多根 K 线——这是预期行为，不是 bug。外汇与贵金属杠杆交易风险极高，星历相关性仅作辅助参考，实盘前务必自行回测。

MQL5 / C++

class="kw">import pandas as pd
# Load data
astro_data = pd.read_csv(&class="macro">#x27;astronomical_data_2018_2024.csv&class="macro">#x27;)
financial_data = pd.read_csv(&class="macro">#x27;EURUSD_data.csv&class="macro">#x27;)
# Convert date columns to class="type">class="kw">datetime
astro_data[&class="macro">#x27;date&class="macro">#x27;] = pd.to_datetime(astro_data[&class="macro">#x27;date&class="macro">#x27;])
financial_data[&class="macro">#x27;time&class="macro">#x27;] = pd.to_datetime(financial_data[&class="macro">#x27;time&class="macro">#x27;])
# Merge data
merged_data = pd.merge(financial_data, astro_data, 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;)
# Save merged data
merged_data.to_csv(&class="macro">#x27;merged_astro_financial_data.csv&class="macro">#x27;, index=False)

◍ 用相关系数扫一遍天文与欧元报价

把前面抓到的行星坐标、月相、太阳活动，和 MT5 拉出的 EURUSD 日线塞进同一张表，就能用 Pandas 的 corr() 直接算全字段相关系数。脚本里对天文列做了 1~5 日滞后偏移，再和 open/high/low/close 等数值列求相关，最后用 seaborn 画出 20×16 英寸的冷热图，并导出与收盘价最显著的几项。实测区间取 2024-03-01 至 2024-07-30，共约 152 个交易日。深入跑完相关地图后，过去天文数据与市场指标之间没露出任何明确相关性，强相关项基本缺席。相关性不等于因果。即便某次算出天文值与价格走势强相关，也只代表统计共生，不意味一方决定另一方。相关地图只是最底层的筛查，真要下结论得另做控制变量研究。外汇与贵金属杠杆高，这类未验证关联绝不能直接当进场依据。

MQL5 / C++

class="kw">import pandas as pd
class="kw">import numpy as np
from skyfield.api class="kw">import load, wgs84, utc
from skyfield.data class="kw">import mpc
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
# Part class="num">1: Collecting astronomical data
# Loading planetary ephemerides
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:
        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
# Collecting astronomical data
start_date = class="type">class="kw">datetime(class="num">2024, class="num">3, class="num">1, tzinfo=utc)
end_date = class="type">class="kw">datetime(class="num">2024, class="num">7, class="num">30, tzinfo=utc)
current_date = start_date
astronomical_data = []
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)
  data = {
    &class="macro">#x27;date&class="macro">#x27;: current_date,
    &class="macro">#x27;moon_phase&class="macro">#x27;: moon_phase,
    &class="macro">#x27;sunspot_number&class="macro">#x27;: solar_activity.get(&class="macro">#x27;sunspot_number&class="macro">#x27;),
    &class="macro">#x27;f10.7_flux&class="macro">#x27;: solar_activity.get(&class="macro">#x27;f10.7_flux&class="macro">#x27;),
    **planet_positions,
    **aspects
  }
  astronomical_data.append(data)
  current_date += timedelta(days=class="num">1)
  print(f"Processed: {current_date}")
# Convert data to DataFrame and save
astro_df = pd.DataFrame(astronomical_data)
astro_df.to_csv(&class="macro">#x27;astronomical_data_2018_2024.csv&class="macro">#x27;, index=False)
print("Astronomical data saved to astronomical_data_2018_2024.csv")
# Part class="num">2: Retrieving financial data via MetaTrader5
# Initialize connection to MetaTrader5
if not mt5.initialize():
  print("initialize() failed")
  mt5.shutdown()
# Set request parameters
symbol = "EURUSD"
timeframe = mt5.TIMEFRAME_D1
start_date = class="type">class="kw">datetime(class="num">2024, class="num">3, class="num">1)
end_date = class="type">class="kw">datetime(class="num">2024, class="num">7, class="num">30)
# Request historical data
rates = mt5.copy_rates_range(symbol, timeframe, start_date, end_date)
# Convert data to DataFrame
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;)
# Save financial data
financial_df.to_csv(f&class="macro">#x27;{symbol}_data.csv&class="macro">#x27;, index=False)
print(f"Financial data saved to {symbol}_data.csv")
# Shutdown MetaTrader5 connection
mt5.shutdown()
# Part class="num">3: Synchronizing astronomical and financial data
# Load data
astro_df = pd.read_csv(&class="macro">#x27;astronomical_data_2018_2024.csv&class="macro">#x27;)
financial_df = pd.read_csv(&class="macro">#x27;EURUSD_data.csv&class="macro">#x27;)
# Convert date columns to class="type">class="kw">datetime
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;])
# Merge data
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;)
# Save merged data
merged_data.to_csv(&class="macro">#x27;merged_astro_financial_data.csv&class="macro">#x27;, index=False)
print("Merged data saved to merged_astro_financial_data.csv")
# Part class="num">4: Statistical analysis of correlations
# Select numeric columns for correlation analysis
numeric_columns = merged_data.select_dtypes(include=[np.number]).columns
# Create lags for astronomical data
for col in numeric_columns:
  if col not 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;, &class="macro">#x27;tick_volume&class="macro">#x27;, &class="macro">#x27;spread&class="macro">#x27;, &class="macro">#x27;real_volume&class="macro">#x27;]:
    for lag in range(class="num">1, class="num">6):  # Create lags from class="num">1 to class="num">5
      merged_data[f&class="macro">#x27;{col}_lag{lag}&class="macro">#x27;] = merged_data[col].shift(lag)
# Update list of numeric columns
numeric_columns = merged_data.select_dtypes(include=[np.number]).columns
# Calculate correlation matrix
correlation_matrix = merged_data[numeric_columns].corr()
# Create heatmap of correlations
plt.figure(figsize=(class="num">20, class="num">16))
sns.heatmap(correlation_matrix, annot=False, cmap=&class="macro">#x27;coolwarm&class="macro">#x27;, vmin=-class="num">1, vmax=class="num">1, center=class="num">0)
plt.title(&class="macro">#x27;Correlation Matrix of Astronomical Factors(with Lags) and EURUSD Prices&class="macro">#x27;)
plt.tight_layout()
plt.savefig(&class="macro">#x27;correlation_heatmap_with_lags.png&class="macro">#x27;)
plt.close()
# Output the most significant correlations with the closing price

把天文滞后项塞进相关性热力图

上面这段把数值列里非 OHLCV 的字段全部挑出来，视作天文因子滞后项，再补一列当前 close，合成 astro_correlation_matrix。用 pandas 的 corr() 默认皮尔逊系数，能量化这些因子与 EURUSD 现价的线性耦合程度，但外汇高杠杆品种里相关系数随时段漂移，结论只代表样本内倾向。 seaborn 的 heatmap 关掉 annot 只留色块，cmap 用 coolwarm 且硬性锁 vmin=-1、vmax=1、center=0，避免颜色被极端值带偏。图幅拉到 18×14、坐标字号 30，是为后期存成 astro_correlation_heatmap_with_lags.png 还能肉眼辨列名。跑完脚本会落 CSV 与 PNG，验证路径直接看工作目录。若你换品种，把 title 里的 EURUSD 改掉即可，但贵金属与外汇均属高风险，滞后相关性高不等于下一根 K 线同向概率高。

MQL5 / C++

significant_correlations = correlation_matrix[&class="macro">#x27;close&class="macro">#x27;].sort_values(key=abs, ascending=False)
print("Most significant correlations with the closing price:")
print(significant_correlations)
# Create a separate correlation matrix for astronomical data with lags and the current price
astro_columns = [col for col in numeric_columns if col not 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;tick_volume&class="macro">#x27;, &class="macro">#x27;spread&class="macro">#x27;, &class="macro">#x27;real_volume&class="macro">#x27;]]
astro_columns.append(&class="macro">#x27;close&class="macro">#x27;)  # Add the current closing price
astro_correlation_matrix = merged_data[astro_columns].corr()
# Create heatmap of correlations for astronomical data with lags and the current price
class="kw">import seaborn as sns
class="kw">import matplotlib.pyplot as plt
# Increase the header and axis label font
plt.figure(figsize=(class="num">18, class="num">14))
sns.heatmap(astro_correlation_matrix, annot=False, cmap=&class="macro">#x27;coolwarm&class="macro">#x27;, vmin=-class="num">1, vmax=class="num">1, center=class="num">0, cbar_kws={&class="macro">#x27;label&class="macro">#x27;: &class="macro">#x27;Correlation&class="macro">#x27;})
plt.title(&class="macro">#x27;Correlation matrix of astronomical factors(with lags) and current EURUSD price&class="macro">#x27;, fontsize=class="num">24)
plt.xlabel(&class="macro">#x27;X-axis Label&class="macro">#x27;, fontsize=class="num">30)
plt.ylabel(&class="macro">#x27;Y-axis Label&class="macro">#x27;, fontsize=class="num">30)
plt.xticks(fontsize=class="num">30)
plt.yticks(fontsize=class="num">30)
plt.tight_layout()
plt.savefig(&class="macro">#x27;astro_correlation_heatmap_with_lags.png&class="macro">#x27;)
plt.close()
print("Analysis completed. Results saved in CSV and PNG files.")

把天文行情对齐交给小布

这类跨源数据清洗和相关矩阵绘制极耗精力，小布盯盘的 AIGC 已内置多源对齐模块，打开 EURUSD 品种页即可一键生成相位叠加视图，你只管读信号。

←

威廉·江恩（William Gann）方法（第三部分）：占星术是否有效？（基础篇）

→

威廉·江恩（William Gann）方法（第三部分）：占星术是否有效？·综合运用

常见问题

在分钟级行情对齐中，Skyfield 的小数点后几位精度通常远超报价跳动周期，误差对相关性结论影响可忽略，重点在时区与历表版本统一。

可用 Python 先算各时刻相位标签，再以时间索引 join 到 MT5 的 OHLC 序列，按相位分组做收益分布比较，避免肉眼看图臆断。

文献倾向将其归为长期趋势扰动项，样本需以月级以上跨度统计，日线噪声会掩盖其微弱效应，外汇贵金属仍属高风险敞口。

可以，内置模块支持天文事件与多品种走势的同步相关分析，省去自己写 Skyfield 与 MT5 API 对接的重复劳动。

多重检验下随机巧合概率随变量数膨胀，须做 FDR 校正或样本外验证，否则行星信号可能只是过拟合产物。

威廉·江恩（William Gann）方法（第三部分）：占星术是否有效？·进阶篇

◍ 星体相位与天文数据的批量抓取逻辑

用 Python 直连 MT5 拉 EURUSD 日线

「把星历表和 EURUSD 拼到同一张表」

◍ 用相关系数扫一遍天文与欧元报价

把天文滞后项塞进相关性热力图

常见问题

相关阅读