查询到15分钟 和5分钟 的合集

import requests
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
import os
import time

# 设置币安API参数
BINANCE_API_URL = "https://api.binance.com/api/v3/klines"
TIMEZONE_OFFSET = 8  # 北京时间UTC+8
RESULT_FILE = "bian_res.txt"

DINGDING_WEBHOOK='https://oapi.dingtalk.com/robot/send?access_token=8a6ddcf98d3b47c63333580bfe9d0bad55b17272eea05cc9c0af7f7be4de070d'
def send_dingding_message(message):
    headers = {"Content-Type": "application/json"}
    data = {
        "msgtype": "text",
        "text": {
            "content": message
        }
    }
    requests.post(DINGDING_WEBHOOK, json=data, headers=headers)

def write_to_file(content, filename=RESULT_FILE):
    """将内容追加写入文件"""
    with open(filename, 'a', encoding='utf-8') as f:
        f.write(content + '\n')


def signal_exists(time_5m, time_15m, filename=RESULT_FILE):
    """检查信号是否已存在于结果文件中"""
    if not os.path.exists(filename):
        return False

    time_5m_str = f"信号时间 (5m): {time_5m}"
    time_15m_str = f"信号时间 (15m): {time_15m}"

    with open(filename, 'r', encoding='utf-8') as f:
        content = f.read()
        return time_5m_str in content and time_15m_str in content


def get_binance_kline(symbol, interval, limit=1000):
    """
    从币安API获取K线数据
    :param symbol: 交易对，如"BTCUSDT"
    :param interval: K线周期，如"5m"或"15m"
    :param limit: 获取的数据条数
    :return: 包含K线数据的DataFrame
    """
    params = {
        'symbol': symbol,
        'interval': interval,
        'limit': limit
    }

    response = requests.get(BINANCE_API_URL, params=params)
    data = response.json()

    if not isinstance(data, list):
        raise Exception(f"API请求失败: {data}")

    # 币安K线数据字段:
    # [开盘时间, 开盘价, 最高价, 最低价, 收盘价, 成交量, 收盘时间, 成交额, 交易数, Taker买入量, Taker买入额, 忽略]
    df = pd.DataFrame(data, columns=[
        'open_time', 'open', 'high', 'low', 'close', 'volume',
        'close_time', 'quote_volume', 'trades', 'taker_buy_volume',
        'taker_buy_quote_volume', 'ignore'
    ])

    # 转换数据类型
    df['timestamp'] = pd.to_datetime(df['open_time'], unit='ms') + timedelta(hours=TIMEZONE_OFFSET)
    for col in ['open', 'high', 'low', 'close', 'volume', 'quote_volume']:
        df[col] = pd.to_numeric(df[col])

    # 按时间排序
    df = df.sort_values('timestamp')

    return df


def calculate_ema(data, period):
    """
    计算指数移动平均线(EMA)
    :param data: 价格数据
    :param period: 计算周期
    :return: EMA值
    """
    return data.ewm(span=period, adjust=False).mean()


def calculate_bollinger_bands(data, period=20, std_dev=2):
    """
    计算布林带
    :param data: 价格数据
    :param period: 计算周期
    :param std_dev: 标准差倍数
    :return: 中轨、上轨、下轨
    """
    sma = data.rolling(window=period).mean()
    std = data.rolling(window=period).std()

    upper_band = sma + (std * std_dev)
    lower_band = sma - (std * std_dev)

    return sma, upper_band, lower_band


def detect_ema_cross(df):
    """
    检测EMA交叉
    :param df: 包含EMA9和EMA26的DataFrame
    :return: 交叉点列表 (方向, 时间)
    """
    crosses = []
    prev_ema9 = df['ema9'].iloc[0]
    prev_ema26 = df['ema26'].iloc[0]

    for i in range(1, len(df)):
        ema9 = df['ema9'].iloc[i]
        ema26 = df['ema26'].iloc[i]

        # 检测向上交叉 (EMA9从下向上穿过EMA26)
        if prev_ema9 < prev_ema26 and ema9 > ema26:
            crosses.append(('up', df['timestamp'].iloc[i]))

        # 检测向下交叉 (EMA9从上向下穿过EMA26)
        if prev_ema9 > prev_ema26 and ema9 < ema26:
            crosses.append(('down', df['timestamp'].iloc[i]))

        prev_ema9 = ema9
        prev_ema26 = ema26

    return crosses


def format_signal(signal, df_5m):
    """格式化信号为字符串"""
    signal_str = f"信号时间 (5m): {signal['time_5m']}\n"
    signal_str += f"信号时间 (15m): {signal['time_15m']}\n"
    signal_str += f"交叉方向: {'向上' if signal['direction'] == 'up' else '向下'}\n"
    signal_str += f"价格: {signal['price']}\n"
    signal_str += f"布林带中轨: {signal['bb_middle']}\n"
    signal_str += f"价格位置: {'中轨之上' if signal['bb_position'] == 'above' else '中轨之下'}\n"
    signal_str += f"预测信号: {signal['signal']}\n"

    # 计算支撑和压力
    cross_time_mask = df_5m['timestamp'] == signal['time_5m']
    if cross_time_mask.any():
        bb_lower = df_5m.loc[cross_time_mask, 'bb_lower'].values[0]
        bb_upper = df_5m.loc[cross_time_mask, 'bb_upper'].values[0]

        signal_str += f"\n支撑和压力线:\n"
        signal_str += f"- 短期支撑: {bb_lower:.2f}\n"
        signal_str += f"- 中期支撑: {signal['bb_middle']:.2f}\n"
        signal_str += f"- 短期压力: {bb_upper:.2f}\n"
    else:
        signal_str += "\n警告: 无法找到交叉时间点的布林带值\n"

    # 根据信号类型给出建议
    if 'bullish' in signal['signal']:
        if signal['bb_position'] == 'above':
            signal_str += "\n预测: 强烈看涨 - 价格在中轨之上且EMA双周期向上交叉\n"
        else:
            signal_str += "\n预测: 谨慎看涨 - 价格在中轨之下但EMA双周期向上交叉\n"
    else:
        if signal['bb_position'] == 'below':
            signal_str += "\n预测: 强烈看跌 - 价格在中轨之下且EMA双周期向下交叉\n"
        else:
            signal_str += "\n预测: 谨慎看跌 - 价格在中轨之上但EMA双周期向下交叉\n"

    signal_str += "=" * 50 + "\n"
    return signal_str


def analyze_trading_signals(symbol):
    """
    分析交易信号
    :param symbol: 交易对，如"BTCUSDT" (币安格式，无连接符)
    """
    # 获取5分钟和15分钟K线数据
    df_5m = get_binance_kline(symbol, '5m')
    df_15m = get_binance_kline(symbol, '15m')

    # 计算技术指标
    for df in [df_5m, df_15m]:
        df['ema9'] = calculate_ema(df['close'], 9)
        df['ema26'] = calculate_ema(df['close'], 26)
        df['bb_middle'], df['bb_upper'], df['bb_lower'] = calculate_bollinger_bands(df['close'])

    # 检测EMA交叉
    crosses_5m = detect_ema_cross(df_5m)
    crosses_15m = detect_ema_cross(df_15m)

    # 寻找符合条件的交叉信号
    new_signals = []

    for cross_15m in crosses_15m:
        direction_15m, time_15m = cross_15m

        # 寻找1小时内同方向的5分钟交叉
        for cross_5m in crosses_5m:
            direction_5m, time_5m = cross_5m

            time_diff = abs((time_5m - time_15m).total_seconds())

            if direction_5m == direction_15m and time_diff <= 90*60:
                # 检查信号是否已存在
                if not signal_exists(time_5m, time_15m):
                    # 找到匹配的交叉，检查布林带位置
                    if direction_5m == 'up':
                        signal_type = 'bullish'
                    else:
                        signal_type = 'bearish'

                    # 获取交叉后的价格和布林带中轨
                    if df_5m['timestamp'].iloc[-1] >= time_5m:
                        idx = df_5m[df_5m['timestamp'] >= time_5m].index[0]
                        if idx < len(df_5m) - 1:
                            price = df_5m['close'].iloc[idx + 1]
                            bb_middle = df_5m['bb_middle'].iloc[idx + 1]

                            bb_position = 'above' if price > bb_middle else 'below'
                            confirmed_signal = ('strong ' if bb_position == 'above' else 'weak ') + signal_type

                            signal = {
                                'time_5m': time_5m,
                                'time_15m': time_15m,
                                'direction': direction_5m,
                                'price': price,
                                'bb_middle': bb_middle,
                                'bb_position': bb_position,
                                'signal': confirmed_signal
                            }
                            new_signals.append(signal)

    # 如果有新信号才写入文件
    if new_signals:
        header = f"\n分析结果 for {symbol} at {datetime.now()}:\n" + "=" * 50 + "\n"
        write_to_file(header)
        # 获取当前时间
        current_time = datetime.now()
        for signal in new_signals:
            if (current_time - signal['time_5m']).total_seconds() <= 2 * 3600:
                signal_str = format_signal(signal, df_5m)
                write_to_file("".format(symbol)+signal_str)

                send_dingding_message("".format({symbol}) + "预测：" +signal_str)

        print(f"发现 {len(new_signals)} 个新信号，已写入 {RESULT_FILE} 文件")
        return True
    else:
        print("没有发现新信号")
        return False


if __name__ == "__main__":
    for i in range(1000000):
        try:
            symbol = "ETHUSDT"  # 币安格式的交易对，无连接符
            analyze_trading_signals(symbol)
        except Exception as e:
            print(f"发生错误: {str(e)}")
        time.sleep(10)