基于 EMA12 指标结合 iTick 外汇报价 API 、股票报价API、指数报价API的量化策略编写与回测

      import requests
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

    

      def get_quote_data(symbol, api_key, start_date, end_date, interval):
    url = f"https://api.itick.com/quote?symbol={symbol}&api_key={api_key}&start_date={start_date}&end_date={end_date}&interval={interval}"
    try:
        response = requests.get(url)
        response.raise_for_status()
        data = response.json()
        df = pd.DataFrame(data['data'])
        df['timestamp'] = pd.to_datetime(df['timestamp'])
        df.set_index('timestamp', inplace=True)
        return df
    except requests.RequestException as e:
        print(f"请求出错: {e}")
        return None
    except KeyError as e:
        print(f"数据解析出错: {e}")
        return None

    

      api_key = "your_api_key"
symbol = "EURUSD"  # 外汇交易对，也可以替换为股票代码或指数代码
start_date = "2023-01-01"
end_date = "2023-12-31"
interval = "1d"  # 数据间隔，这里设置为日线数据

data = get_quote_data(symbol, api_key, start_date, end_date, interval)
if data is not None:
    print(data.head())

    

      def calculate_ema12(data):
    data['ema12'] = data['close'].ewm(span=12, adjust=False).mean()
    return data

    

      data = calculate_ema12(data)
if data is not None:
    print(data[['close', 'ema12']].head())

    

      def generate_signals(data):
    data['signal'] = 0
    data.loc[data['close'] > data['ema12'], 'signal'] = 1  # 买入信号
    data.loc[data['close'] < data['ema12'], 'signal'] = -1  # 卖出信号
    data['position'] = data['signal'].diff()
    return data

    

      data = generate_signals(data)
if data is not None:
    print(data[['close', 'ema12', 'signal', 'position']].head())

    

      def backtest(data):
    initial_capital = float(100000.0)
    positions = pd.DataFrame(index=data.index).fillna(0.0)
    positions[symbol] = 100 * data['signal']  # 假设每次交易 100 单位

    portfolio = positions.multiply(data['close'], axis=0)
    pos_diff = positions.diff()

    portfolio['holdings'] = (positions.multiply(data['close'], axis=0)).sum(axis=1)
    portfolio['cash'] = initial_capital - (pos_diff.multiply(data['close'], axis=0)).sum(axis=1).cumsum()
    portfolio['total'] = portfolio['cash'] + portfolio['holdings']
    portfolio['returns'] = portfolio['total'].pct_change()

    return portfolio

    

      portfolio = backtest(data)
if portfolio is not None:
    print(portfolio[['holdings', 'cash', 'total', 'returns']].head())

    

      if data is not None:
    plt.figure(figsize=(12, 6))
    plt.plot(data['close'], label='Close Price')
    plt.plot(data['ema12'], label='EMA12')
    plt.title(f'{symbol} Close Price and EMA12')
    plt.xlabel('Date')
    plt.ylabel('Price')
    plt.legend()
    plt.show()

    

      if portfolio is not None:
    plt.figure(figsize=(12, 6))
    plt.plot(portfolio['total'], label='Portfolio Value')
    plt.title('Portfolio Value over Time')
    plt.xlabel('Date')
    plt.ylabel('Value')
    plt.legend()
    plt.show()