Python教程第8节: 如何计算历史杠杆ETF价格？

很多朋友问我，我回测中的历史杠杆ETF的数据是怎么来的，今天我就告诉大家如何根据股票历史价格计算出相应的杠杆ETF价格。

杠杆ETF运行机制

首先我们得明白杠杆ETF的运行机制，杠杆ETF每天结算，每天的涨跌幅等于没有杠杆的涨幅或者跌幅乘以杠杆比例，然后还要扣除ETF管理费用。

计算变化率函数

我们先写一个函数计算原始股票价格每天的变化率。

def _compute_change_ratio(previous_value: float,
                          current_value: float
                          ):
    change_ratio = current_value / previous_value
    return change_ratio

杠杆ETF价格变化函数

接下来，我们需要编写一个函数来计算杠杆ETF的价格变化。考虑到管理费用，我们还需引入一个修正值。

def _compute_leveraged_change_ratio(previous_value: float,
                                    current_value: float,
                                    leverage: int,
                                    correction_rate: float):
    change_ratio = _compute_change_ratio(previous_value, current_value)
    leveraged_changed_ratio = (1 + (change_ratio - 1) * leverage) * correction_rate
    return leveraged_changed_ratio

计算开盘价的杠杆ETF价格

我们先写一个初始版本，只计算带杠杆的开盘价，给3个初始变量：

杠杆比例leverage=2
每日价格修正值先设为1（daily_correction_rate=1），也就是不修正
杠杆ETF第一天的股价设为1（init_value=1），这个值等于多少无所谓。，并展示其效果。

def main():
    prices = get_prices("^NDX")
    leverage = 2
    daily_correction_rate = 1 ** (1 / 252)
    init_value = 1

    leveraged_opens = [init_value]
    for price_id in range(1, len(prices)):
        leveraged_change_ratio_open = _compute_leveraged_change_ratio(
            prices.iloc[price_id - 1]["Open"],
            prices.iloc[price_id]["Open"],
            leverage,
            daily_correction_rate)
        new_leveraged_open = leveraged_opens[-1] * leveraged_change_ratio_open
        leveraged_opens.append(new_leveraged_open)
    leveraged_prices = DataFrame({"Open": leveraged_opens,
                                  "Close": leveraged_opens,
                                  "High": leveraged_opens,
                                  "Low": leveraged_opens,
                                  'Adj Close': leveraged_opens}, index=prices.index)
    plot_in_chart(leveraged_prices, 50)


if __name__ == '__main__':
    main()

计算全部价格并封装函数

接下来，我们将补充收盘价、最高价和最低价的计算，并将代码封装成一个函数。

PS：这里我加入了函数变量的type信息（DataFrame，int，float），这在Python中不是必须的，但是加入变量内容信息更容易维护代码。

def compute_leveraged_prices(prices: DataFrame,
                             leverage: int,
                             init_value: float,
                             daily_correction_rate: float = 1):
    leveraged_opens = [init_value]
    leveraged_closes = [init_value * _compute_leveraged_change_ratio(
        prices.iloc[0]["Open"],
        prices.iloc[0]["Close"],
        leverage)]
    leveraged_highs = [init_value * _compute_leveraged_change_ratio(
        prices.iloc[0]["Open"],
        prices.iloc[0]["High"],
        leverage)]
    leveraged_lows = [init_value * _compute_leveraged_change_ratio(
        prices.iloc[0]["Open"],
        prices.iloc[0]["Low"],
        leverage)]

    for price_id in range(1, len(prices)):
        leveraged_change_ratio_open = _compute_leveraged_change_ratio(
            prices.iloc[price_id - 1]["Open"],
            prices.iloc[price_id]["Open"],
            leverage,
            daily_correction_rate)
        new_leveraged_open = leveraged_opens[-1] * leveraged_change_ratio_open
        leveraged_opens.append(new_leveraged_open)
        leveraged_closes.append(new_leveraged_open * _compute_leveraged_change_ratio(
            prices.iloc[price_id]["Open"],
            prices.iloc[price_id]["Close"],
            leverage))
        leveraged_highs.append(new_leveraged_open * _compute_leveraged_change_ratio(
            prices.iloc[price_id]["Open"],
            prices.iloc[price_id]["High"],
            leverage))
        leveraged_lows.append(new_leveraged_open * _compute_leveraged_change_ratio(
            prices.iloc[price_id]["Open"],
            prices.iloc[price_id]["Low"],
            leverage))

    leveraged_prices = DataFrame({"Open": leveraged_opens,
                                  "Close": leveraged_closes,
                                  "High": leveraged_highs,
                                  "Low": leveraged_lows,
                                  'Adj Close': leveraged_closes}, index=prices.index)
    return leveraged_prices

改进并调试

我们可以根据实际情况调整管理费用等参数，使得计算结果更加准确。

我们可以使用QLD官方给出的的年化0.95%的管理费作为参考，按照一年252个交易日计算，每个交易日的损失率就是(1-0.0095)**(1/252)，修正比率也就是1-0.0095**(1/252)，最后的代码变成这样。我们也可以把这里的函数全部放到别的py文件中，比如我们在前几节中建的utils.py。

# main.py

from pandas import DataFrame

from utils import get_prices, plot_in_chart


def _compute_change_ratio(previous_value: float,
                          current_value: float
                          ):
    change_ratio = current_value / previous_value
    return change_ratio


def _compute_leveraged_change_ratio(previous_value: float,
                                    current_value: float,
                                    leverage: int,
                                    correction_rate: float = 1):
    change_ratio = _compute_change_ratio(previous_value, current_value)
    leveraged_changed_ratio = (1 + (change_ratio - 1) * leverage) * correction_rate
    return leveraged_changed_ratio


def compute_leveraged_prices(prices: DataFrame,
                             leverage: int,
                             init_value: float,
                             daily_correction_rate: float = 1):
    leveraged_opens = [init_value]
    leveraged_closes = [init_value * _compute_leveraged_change_ratio(
        prices.iloc[0]["Open"],
        prices.iloc[0]["Close"],
        leverage)]
    leveraged_highs = [init_value * _compute_leveraged_change_ratio(
        prices.iloc[0]["Open"],
        prices.iloc[0]["High"],
        leverage)]
    leveraged_lows = [init_value * _compute_leveraged_change_ratio(
        prices.iloc[0]["Open"],
        prices.iloc[0]["Low"],
        leverage)]

    for price_id in range(1, len(prices)):
        leveraged_change_ratio_open = _compute_leveraged_change_ratio(
            prices.iloc[price_id - 1]["Open"],
            prices.iloc[price_id]["Open"],
            leverage,
            daily_correction_rate)
        new_leveraged_open = leveraged_opens[-1] * leveraged_change_ratio_open
        leveraged_opens.append(new_leveraged_open)
        leveraged_closes.append(new_leveraged_open * _compute_leveraged_change_ratio(
            prices.iloc[price_id]["Open"],
            prices.iloc[price_id]["Close"],
            leverage))
        leveraged_highs.append(new_leveraged_open * _compute_leveraged_change_ratio(
            prices.iloc[price_id]["Open"],
            prices.iloc[price_id]["High"],
            leverage))
        leveraged_lows.append(new_leveraged_open * _compute_leveraged_change_ratio(
            prices.iloc[price_id]["Open"],
            prices.iloc[price_id]["Low"],
            leverage))

    leveraged_prices = DataFrame({"Open": leveraged_opens,
                                  "Close": leveraged_closes,
                                  "High": leveraged_highs,
                                  "Low": leveraged_lows,
                                  'Adj Close': leveraged_closes}, index=prices.index)
    return leveraged_prices


def main():
    prices = get_prices("^NDX")
    leveraged_prices = compute_leveraged_prices(prices,
                                                2,
                                                1,
                                                (1-0.0095)**(1/252))
    plot_in_chart(leveraged_prices, 50)


if __name__ == '__main__':
    main()

# utils.py

import warnings
import pandas as pd
import yfinance as yf
from lightweight_charts import Chart


def get_prices(stock_symbol, interval="1d"):
    # 1wk for 1 week, 1mo for 1 month
    data = yf.download(stock_symbol, interval=interval)
    return data


def get_moving_average(prices, window_size, mode="sma"):
    if mode == "sma":
        sma = prices['Adj Close'].rolling(window=window_size).mean()
        return sma

    elif mode == "ema":
        ema = prices['Adj Close'].emw(span=window_size).mean()
        return ema
    else:
        warnings.warn(f"{mode} is not a known mode!!")


def plot_in_chart(prices, ma_windows_size, ma_mode="sma"):
    chart = Chart()
    chart.set(prices)
    chart.legend(visible=True)
    column_name = f'{ma_mode} {ma_windows_size}'.upper()
    line = chart.create_line(column_name)
    sma = pd.DataFrame({"Date": prices.index,
                        column_name: get_moving_average(prices, ma_windows_size, ma_mode)})
    line.set(sma)
    chart.show(block=True)

最终得到的结果可以与QLD的实际价格进行对比，进行微调以使计算结果更加准确。同时，也可以根据需求自行计算不同杠杆倍数的ETF价格，根据我的经验，杠杆倍数越高，修正值越小，损耗也越严重，这也是为什么杠杆ETF长期持有风险大的原因。

保存导出数据

当然，你也可以选择把计算出来的数据保存下来，这也很简单

    leveraged_prices.to_csv("your_path.csv")

5 1 vote

Article Rating