AHR999监控脚本

ahr999_monitor.py

import math
import time
from datetime import date, datetime, timedelta
from urllib.parse import quote
from zoneinfo import ZoneInfo

import requests

# ================= 配置区 =================
# 美股开盘前 N 分钟执行（美股 9:30 开盘，默认提前 10 分钟 = 9:20 美东时间）
RUN_HOUR, RUN_MINUTE = 9, 20
RUN_TIMEZONE = "America/New_York"

# 比特币创世区块时间 (用于计算比特币的“币龄”)
GENESIS_DATE = date(2009, 1, 3)

# 拟合估值公式参数：币价 = 10^(A * log10(币龄) - B)
# 原始版 (2019): A=5.84, B=17.01
# 更新版 (2020.4+): A=5.8, B=16.88，拟合更贴近近期数据
FIT_A, FIT_B = 5.8, 16.88

# Bark 推送配置 (https://github.com/Finb/Bark)
BARK_SERVER = "http://your_service_ip:your_service_port/your_token"
# ==========================================


def get_bitcoin_data_from_binance():
    """
    从币安 (Binance) 免费公开 API 获取过去 200 天的比特币每日收盘价
    """
    url = "https://api.binance.com/api/v3/klines"
    params = {"symbol": "BTCUSDT", "interval": "1d", "limit": 200}

    response = requests.get(url, params=params, timeout=10)
    response.raise_for_status()
    data = response.json()

    # 币安 Kline 格式:[Open time, Open, High, Low, Close, Volume, ...]
    # 索引 4 是每日收盘价
    closes = [float(kline[4]) for kline in data]
    return closes


def get_bitcoin_data_from_binance_us():
    """
    从币安 (Binance) 免费公开 API 获取过去 200 天的比特币每日收盘价
    """
    url = "https://api.binance.us/api/v3/klines"
    params = {"symbol": "BTCUSDT", "interval": "1d", "limit": 200}

    response = requests.get(url, params=params, timeout=10)
    response.raise_for_status()
    data = response.json()

    # 币安 Kline 格式:[Open time, Open, High, Low, Close, Volume, ...]
    # 索引 4 是每日收盘价
    closes = [float(kline[4]) for kline in data]
    return closes


def get_bitcoin_data_from_kraken():
    """
    改用 Kraken (海妖) 交易所公开 API 获取比特币日线
    无地区 451 限制
    """
    # 1440分钟 = 1天
    url = "https://api.kraken.com/0/public/OHLC"
    params = {
        "pair": "XBTUSD",  # Kraken 中 BTC 代码为 XBT
        "interval": 1440,
    }

    response = requests.get(url, params=params, timeout=10)
    response.raise_for_status()
    data = response.json()

    if data.get("error"):
        raise Exception(f"Kraken API Error: {data['error']}")

    # Kraken 数据结构: data['result']['XXBTZUSD'] 是一个列表
    # 每一根 K 线的格式:[time, open, high, low, close, vwap, volume, count]
    # 索引 4 为每日收盘价
    klines = data["result"]["XXBTZUSD"]

    # 取最近的 200 天数据
    closes = [float(kline[4]) for kline in klines[-200:]]
    return closes


def calculate_ahr999(closes):
    """
    计算 AHR999 指数
    公式: AHR999 = (当前价格 / 200日定投成本) * (当前价格 / 拟合估值)
    """
    # 过滤零值/负值，避免 math.log(0) 报错
    closes = [p for p in closes if p and p > 0]
    if len(closes) < 200:
        raise ValueError("有效价格数据不足 200 天（存在 0 或负值）")

    current_price = closes[-1]

    # 1. 计算200日定投成本 (200日价格的几何平均数)
    # 几何平均数公式: exp(sum(ln(price)) / N)
    gma200 = math.exp(sum(math.log(price) for price in closes) / len(closes))

    # 2. 计算比特币拟合估值 (基于历史的指数增长模型估值)
    age_days = (date.today() - GENESIS_DATE).days
    estimated_price = 10 ** (FIT_A * math.log10(age_days) - FIT_B)

    # 3. 计算 AHR999 指数
    ahr999 = (current_price / gma200) * (current_price / estimated_price)

    return ahr999, current_price, gma200, estimated_price


def send_notification(message):
    """
    消息通知模块：控制台输出 + Bark 推送到 iPhone
    """
    current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    title = "AHR999 监控报告"
    print("-" * 50)
    print(f"🔔 [{current_time}] {title}\n")
    print(message)
    print("-" * 50)

    # Bark 推送 (https://github.com/Finb/Bark)
    if BARK_SERVER:
        try:
            # Bark 格式: /key/title/body
            push_url = f"{BARK_SERVER.rstrip('/')}/{quote(title)}/{quote(message)}"
            resp = requests.get(push_url, timeout=10)
            if resp.status_code == 200:
                print("📱 已推送到 Bark")
            else:
                print(f"⚠️ Bark 推送异常: HTTP {resp.status_code}")
        except Exception as e:
            print(f"⚠️ Bark 推送失败: {e}")


def _next_run_time() -> datetime:
    """计算下次执行时间（美股开盘前 10 分钟，美东时间）"""
    tz = ZoneInfo(RUN_TIMEZONE)
    now = datetime.now(tz)
    today_run = now.replace(hour=RUN_HOUR, minute=RUN_MINUTE, second=0, microsecond=0)
    if now >= today_run:
        today_run += timedelta(days=1)
    return today_run


def _run_once():
    """执行一次 AHR999 检查并推送"""
    try:
        closes = get_bitcoin_data_from_kraken()
        if len(closes) < 200:
            print("⚠️ 获取的 K线数据不足 200 天，API 可能受限。")
            return

        ahr999, current_price, gma200, estimated_price = calculate_ahr999(closes)

        if ahr999 < 0.45:
            strategy = "🟢 抄底区间！价格严重被低估，极佳的买入机会。"
        elif ahr999 <= 1.2:
            strategy = "🟡 定投区间。价格在合理范围内，适合按纪律坚持定投。"
        elif ahr999 <= 5.0:
            strategy = "🟠 观望区间。价格偏高，建议暂停买入，持有观望为主。"
        else:
            strategy = "🔴 逃顶区间！价格处于极度泡沫期，建议分批获利了结。"

        message = (
            f"💰 当前 BTC 价格: ${current_price:,.2f}\n"
            f"📈 200日定投成本: ${gma200:,.2f}\n"
            f"🔮 拟合增长估值: ${estimated_price:,.2f}\n"
            f"🔥 AHR999 指数:  {ahr999:.4f}\n\n"
            f"💡 操作建议: {strategy}"
        )
        send_notification(message)

    except Exception as e:
        print(f"❌ 监控发生异常: {e}")


def run_monitor():
    """每天美股开盘前 10 分钟执行一次"""
    tz = ZoneInfo(RUN_TIMEZONE)
    print(f"🚀 AHR999 每日监控（美东 {RUN_HOUR}:{RUN_MINUTE:02d} 执行）")
    while True:
        next_run = _next_run_time()
        wait_secs = (next_run - datetime.now(tz)).total_seconds()
        print(f"⏰ 下次执行: {next_run.strftime('%Y-%m-%d %H:%M')} 美东（约 {wait_secs / 3600:.1f} 小时后）")
        time.sleep(wait_secs)
        _run_once()


if __name__ == "__main__":
    run_monitor()