支撑与阻力位+钉钉

#直接查看阻力范围，
import time
import hmac
import hashlib
import base64
import urllib.parse
import requests
import json
import pandas as pd


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))
    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))
    print(response.text)


def read_events_from_file(filename="压力线.txt"):
    try:
        with open(filename, "r") as file:
            return [line.strip() for line in file.readlines()]
    except FileNotFoundError:
        return []


def save_events_to_file(events, filename="压力线.txt"):
    with open(filename, "w") as file:
        for event in events:
            file.write(str(event) + "\n")


def append_events_to_file(events, filename="压力线.txt"):
    with open(filename, "a") as file:
        for event in events:
            file.write(str(event) + "\n")


def fetch_klines(symbol, interval, limit=500):
    url = "https://fapi.binance.com/fapi/v1/klines"
    params = {
        'symbol': symbol,
        'interval': interval,
        'limit': limit
    }
    response = requests.get(url, params=params)
    data = response.json()
    return pd.DataFrame(data, columns=['Open Time', 'Open', 'High', 'Low', 'Close', 'Volume',
                                       'Close Time', 'Quote Asset Volume', 'Number of Trades',
                                       'Taker Buy Base Asset Volume', 'Taker Buy Quote Asset Volume', 'Ignore'])


def calculate_gann_levels(df, price_level=3):
    close_prices = df['Close'].astype(float)
    max_val = 20
    gann_numbers = []

    for min in range(max_val + 1):
        for i in range(4):
            gNum = 0
            if min == 0 and i == 0:
                gNum = min + (min + 2)
            elif min > 0 and i == 0:
                gNum = round(gann_numbers[-1]) + (min + 1) + min
            else:
                gNum = round(gann_numbers[-1]) + (min + 2) + min
            gann_numbers.append(gNum)

    last_close = close_prices.iloc[-1]
    if last_close >= 10000:
        denominator = 0.01
    elif last_close >= 1000:
        denominator = 0.1
    elif last_close >= 100:
        denominator = 1
    elif last_close >= 10:
        denominator = 10
    elif last_close >= 0.05:
        denominator = 100
    else:
        denominator = 1000

    price = last_close * denominator
    resistance = 0.0
    support = 0.0

    for i in range(len(gann_numbers) - 1):
        if gann_numbers[i] <= price < gann_numbers[i + 1]:
            resistance = gann_numbers[i + 1] / denominator
            support = gann_numbers[i] / denominator
            break

    blue_gann_price = (support + resistance) / 2

    return support, resistance, blue_gann_price


def main():
    webhook_url = "https://oapi.dingtalk.com/robot/send?access_token=8a6ddcf98d3b47c63333580bfe9d0bad55b17272eea05cc9c0af7f7be4de070d"
    secret = "你的秘钥"
    symbol = "BTCUSDT"
    interval = "1h"
    price_level = 3  # Change as needed (3 or 5)

    # Fetch Kline data
    df = fetch_klines(symbol, interval)

    # Calculate Gann levels
    support, resistance, blue_gann_price = calculate_gann_levels(df, price_level)

    # Store results in a dictionary
    current_results = {
        'support': support,
        'resistance': resistance,
        'blue_gann_price': blue_gann_price
    }

    # Read previous results from file (if it exists)
    previous_results = read_events_from_file()
    if previous_results:
        previous_results = json.loads(previous_results[0])  # Assuming only one line with JSON data
    else:
        previous_results = {}

    # Check if there are changes
    if (previous_results.get('support') != current_results['support'] or
            previous_results.get('resistance') != current_results['resistance'] or
            previous_results.get('blue_gann_price') != current_results['blue_gann_price']):

        # Send message to DingTalk
        message = (
            f"最低支撑价格: {current_results['support']:.2f}\n"
            f"阻力位: {current_results['resistance']:.2f}\n"
            f"当价格在 {current_results['blue_gann_price']:.2f} 之上时，可能表示市场处于上涨趋势，而当价格在其下方时，可能表示下跌趋势,穿越: {current_results['blue_gann_price']:.2f}"
        )
        send_dingtalk_message(webhook_url, secret, message)  # Replace with your actual webhook and secret

        # Save the current results to file
        save_events_to_file([json.dumps(current_results)])
    else:
        # Just print the results
        print(f"最低支撑价格: {support:.2f}")
        print(f"阻力位: {resistance:.2f}")
        print(
            f"当价格在 {blue_gann_price:.2f} 之上时，可能表示市场处于上涨趋势，而当价格在其下方时，可能表示下跌趋势: {blue_gann_price:.2f}")


if __name__ == "__main__":
    while True:
        main()
        time.sleep(60)