量化纏論技術指標與股票實時報價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