获取3分钟金叉和死叉

import time
import hmac
import hashlib
import base64
import urllib.parse
import json
import requests
import pandas as pd


def get_klines(symbol, interval, limit=500):
    """获取K线数据"""
    url = "https://fapi.binance.com/fapi/v1/klines"
    params = {
        'symbol': symbol,
        'interval': interval,
        'limit': limit,
    }
    response = requests.get(url, params=params)
    return response.json()


def calculate_macd(data, short_period=6, long_period=7, signal_period=4):
    """计算MACD指标"""
    # 将获取的数据转换为DataFrame
    df = pd.DataFrame(data, columns=['OpenTime', 'Open', 'High', 'Low', 'Close', 'Volume', 'CloseTime',
                                     'QuoteAssetVolume', 'NumberOfTrades', 'TakerBuyBaseAssetVolume',
                                     'TakerBuyQuoteAssetVolume', 'Ignore'])

    # 将收盘价转换为数值类型
    df['Close'] = pd.to_numeric(df['Close'])

    # 计算快速EMA和慢速EMA
    df['EMA_fast'] = df['Close'].ewm(span=short_period, adjust=False).mean()
    df['EMA_slow'] = df['Close'].ewm(span=long_period, adjust=False).mean()

    # 计算DIF
    df['DIF'] = df['EMA_fast'] - df['EMA_slow']

    # 计算DEA
    df['DEA'] = df['DIF'].ewm(span=signal_period, adjust=False).mean()

    # 计算MACD
    df['MACD'] = (df['DIF'] - df['DEA']) * 2
    return df


def find_macd_crossovers(df):
    """找到MACD的金叉和死叉"""
    crossover_points = {'上穿': [], '下穿': []}
    for i in range(1, len(df)):
        if df['DIF'].iloc[i] > 0 and df['DIF'].iloc[i - 1] <= 0:
            crossover_points['上穿'].append(df['OpenTime'].iloc[i])
        elif df['DIF'].iloc[i] < 0 and df['DIF'].iloc[i - 1] >= 0:
            crossover_points['下穿'].append(df['OpenTime'].iloc[i])
    return crossover_points


def send_dingtalk_message(webhook, secret, message):
    # 获取时间戳
    timestamp = str(round(time.time() * 1000))
    # 构造签名
    secret_enc = secret.encode('utf-8')
    string_to_sign = '{}\n{}'.format(timestamp, secret)
    string_to_sign_enc = string_to_sign.encode('utf-8')
    hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
    sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))
    # 构造请求URL
    webhook = '{}&timestamp={}&sign={}'.format(webhook, timestamp, sign)

    # 构造消息体
    headers = {'Content-Type': 'application/json'}
    data = {
        "msgtype": "text",
        "text": {
            "content": message
        }
    }

    # 发送请求
    response = requests.post(url=webhook, headers=headers, data=json.dumps(data))
    return response.text


# 初始化最后已知的金叉和死叉时间点
last_golden_cross = pd.Timestamp('2024-08-26 17:05:00+08:00')
last_death_cross = pd.Timestamp('2024-08-26 18:08:00+08:00')

webhook_url = "https://oapi.dingtalk.com/robot/send?access_token=8a6ddcf98d3b47c63333580bfe9d0bad55b17272eea05cc9c0af7f7be4de070d"
secret_key = "你的加签密钥"  # 如果没有设置加签，则这部分可以忽略
while True:  # 开始一个死循环
    try:
        # 获取K线数据
        symbol = 'BTCUSDT'  # 示例: BTC/USDT
        interval = '5m'  # 示例: 3分钟
        data = get_klines(symbol, interval)

        # 计算MACD
        df_macd = calculate_macd(data)

        # 找到金叉和死叉
        crossovers = find_macd_crossovers(df_macd)

        # 检查并打印新的金叉和死叉的时间点
        new_crossovers = False
        for cross_type, times in crossovers.items():
            for time_point in times:
                time_utc8 = pd.to_datetime(time_point, unit='ms').tz_localize('UTC').tz_convert('Asia/Shanghai')

                # 检查时间点是否新
                if cross_type == '上穿' and time_utc8 > last_golden_cross:
                    message = f"{interval}穿越，新的{cross_type} at: {time_utc8}"
                    print(message)
                    result = send_dingtalk_message(webhook_url, secret_key, message)
                    print(result)
                    last_golden_cross = time_utc8
                    new_crossovers = True
                elif cross_type == '下穿' and time_utc8 > last_death_cross:
                    message = f"{interval}穿越，新的{cross_type} at: {time_utc8}"
                    print(message)
                    result = send_dingtalk_message(webhook_url, secret_key, message)
                    last_death_cross = time_utc8
                    new_crossovers = True

        if not new_crossovers:
            print("没有新的金叉或死叉。")

        print("Waiting for the next update...")
        time.sleep(2)  # 每10秒查询一次，可以根据需要调整

    except KeyboardInterrupt:
        print("Program terminated by user.")
        break  # 当用户中断程序(通常是通过Ctrl+C)时，退出循环
    except Exception as e:
        print(f"An error occurred: {e}")
        # 出现错误时等待一段时间再重试，避免因临时的网络API限制导致程序终止
        time.sleep(300)