量化缠论技术指标与股票实时报价API结合实现策略交易

      import requests
import json

# iTick API相关配置
API_KEY = "your_api_key"
BASE_URL = "https://api.itick.org"
headers = {
        "accept": "application/json"
        "token": API_KEY
        }

def get_historical_data(symbol, region, endDate):
    """获取历史数据"""
    url = f"{BASE_URL}/stock/kline"
   
    params = {
        "code": symbol,
        "region": region,
        "et":endDate,
        "kType":2,
        "limit":100
    }
    response = requests.get(url, params=params, headers=headers)
    data = json.loads(response.text)
    return data

def get_realtime_quote(symbol,region):
    """获取实时报价数据"""
    url = f"{BASE_URL}/stock/quote"
    params = {
        "region": region,
        "code": symbol
    }
    response = requests.get(url, params=params, headers=headers)
    data = json.loads(response.text)
    return data

# 示例：获取某股票的历史数据和实时报价
symbol = "AAPL"
region = "US"
historical_data = get_historical_data(symbol,region,1755098012000)
realtime_quote = get_realtime_quote(symbol,region)

    

      def find_fractals(data):
    """识别顶分型和底分型"""
    fractals = []
    for i in range(2, len(data) - 2):
        # 顶分型：中间K线最高价高于两侧相邻K线最高价
        if data[i]['high'] > data[i-1]['high'] and data[i]['high'] > data[i-2]['high'] and data[i]['high'] > data[i+1]['high'] and data[i]['high'] > data[i+2]['high']:
            fractals.append({
                'index': i,
                'type': 'top',
                'price': data[i]['high']
            })
        # 底分型：中间K线最低价低于两侧相邻K线最低价
        elif data[i]['low'] < data[i-1]['low'] and data[i]['low'] < data[i-2]['low'] and data[i]['low'] < data[i+1]['low'] and data[i]['low'] < data[i+2]['low']:
            fractals.append({
                'index': i,
                'type': 'bottom',
                'price': data[i]['low']
            })
    return fractals

    

      import numpy as np

def calculate_force(data, start, end):
    """计算某段走势的力度"""
    # 此处简化处理，实际可根据MACD等指标计算
    prices = [d['close'] for d in data[start:end+1]]
    return np.abs(np.diff(prices)).sum()

def judge_divergence(data, segment1_start, segment1_end, segment2_start, segment2_end):
    """判断两段走势是否出现背驰"""
    force1 = calculate_force(data, segment1_start, segment1_end)
    force2 = calculate_force(data, segment2_start, segment2_end)
    # 价格创新高/新低但力度减弱，出现背驰
    price1_high = max([d['high'] for d in data[segment1_start:segment1_end+1]])
    price2_high = max([d['high'] for d in data[segment2_start:segment2_end+1]])
    if price2_high > price1_high and force2 < force1:
        return 'top_divergence'  # 顶背驰
    price1_low = min([d['low'] for d in data[segment1_start:segment1_end+1]])
    price2_low = min([d['low'] for d in data[segment2_start:segment2_end+1]])
    if price2_low < price1_low and force2 < force1:
        return 'bottom_divergence'  # 底背驰
    return None