《python数字货币量化交易》学习笔记 第四部分别有洞天篇 4.1 择时策略实盘与优化
.
4.1.1 策略评价
4.1.2 资金曲线图 Statistics
.
.
.

《python数字货币量化交易》学习笔记 第四部分别有洞天篇 4.1 择时策略实盘与优化 4.1.1 策略评价

“”"

本节课程内容

评价策略好坏的主流指标
“”"
import pandas as pd
from Statistics import *
pd.set_option(‘expand_frame_repr’, False) # 当列太多时不换行
pd.set_option(‘display.max_rows’, 5000) # 最多显示数据的行数

读取资金曲线数据

equity_curve = pd.read_pickle(’/Users/xingbuxingx/Desktop/数字货币量化课程/2020版数字货币量化投资课程/xbx_coin_2020/data/cls-4.1.1/equity_curve.pkl’)

print(equity_curve)

计算每笔交易

trade = transfer_equity_curve_to_trade(equity_curve)
print(trade)

计算各类统计指标

r, monthly_return = strategy_evaluate(equity_curve, trade)

print®
print(monthly_return)

**《python数字货币量化交易》学习笔记  第四部分别有洞天篇   4.1 择时策略实盘与优化  4.1.1 策略评价**

"""
# 本节课程内容
评价策略好坏的主流指标
"""
import pandas as pd
from Statistics import *
pd.set_option('expand_frame_repr', False)  # 当列太多时不换行
pd.set_option('display.max_rows', 5000)  # 最多显示数据的行数


# 读取资金曲线数据
equity_curve = pd.read_pickle('/Users/xingbuxingx/Desktop/数字货币量化课程/2020版数字货币量化投资课程/xbx_coin_2020/data/cls-4.1.1/equity_curve.pkl')
# print(equity_curve)


# 计算每笔交易
trade = transfer_equity_curve_to_trade(equity_curve)
print(trade)


# 计算各类统计指标
r, monthly_return = strategy_evaluate(equity_curve, trade)

print(r)
print(monthly_return)

《python数字货币量化交易》学习笔记 第四部分别有洞天篇 4.1 择时策略实盘与优化 4.1.2 资金曲线图 Statistics

“”"

课程内容

策略评价函数
“”"
import pandas as pd
import numpy as np
import itertools

======= 策略评价 =========

将资金曲线数据，转化为交易数据

def transfer_equity_curve_to_trade(equity_curve):
“”"
将资金曲线数据，转化为一笔一笔的交易
:param equity_curve: 资金曲线函数计算好的结果，必须包含pos
:return:
“”"
# =选取开仓、平仓条件
condition1 = equity_curve[‘pos’] != 0
condition2 = equity_curve[‘pos’] != equity_curve[‘pos’].shift(1)
open_pos_condition = condition1 & condition2

# =计算每笔交易的start_time
if 'start_time' not in equity_curve.columns:
    equity_curve.loc[open_pos_condition, 'start_time'] = equity_curve['candle_begin_time']
    equity_curve['start_time'].fillna(method='ffill', inplace=True)
    equity_curve.loc[equity_curve['pos'] == 0, 'start_time'] = pd.NaT

# =对每次交易进行分组，遍历每笔交易
trade = pd.DataFrame()  # 计算结果放在trade变量中

for _index, group in equity_curve.groupby('start_time'):

    # 记录每笔交易
    # 本次交易方向
    trade.loc[_index, 'signal'] = group['pos'].iloc[0]

    # 本次交易杠杆倍数
    if 'leverage_rate' in group:
        trade.loc[_index, 'leverage_rate'] = group['leverage_rate'].iloc[0]

    g = group[group['pos'] != 0]  # 去除pos=0的行
    # 本次交易结束那根K线的开始时间
    trade.loc[_index, 'end_bar'] = g.iloc[-1]['candle_begin_time']
    # 开仓价格
    trade.loc[_index, 'start_price'] = g.iloc[0]['open']
    # 平仓信号的价格
    trade.loc[_index, 'end_price'] = g.iloc[-1]['close']
    # 持仓k线数量
    trade.loc[_index, 'bar_num'] = g.shape[0]
    # 本次交易收益
    trade.loc[_index, 'change'] = (group['equity_change'] + 1).prod() - 1
    # 本次交易结束时资金曲线
    trade.loc[_index, 'end_equity_curve'] = g.iloc[-1]['equity_curve']
    # 本次交易中资金曲线最低值
    trade.loc[_index, 'min_equity_curve'] = g['equity_curve'].min()

return trade

计算策略评价指标

def strategy_evaluate(equity_curve, trade):
“”"
:param equity_curve: 带资金曲线的df
:param trade: transfer_equity_curve_to_trade的输出结果，每笔交易的df
:return:
“”"

# ===新建一个dataframe保存回测指标
results = pd.DataFrame()

# ===计算累积净值
results.loc[0, '累积净值'] = round(equity_curve['equity_curve'].iloc[-1], 2)

# ===计算年化收益
annual_return = (equity_curve['equity_curve'].iloc[-1] / equity_curve['equity_curve'].iloc[0]) ** (
    '1 days 00:00:00' / (equity_curve['candle_begin_time'].iloc[-1] - equity_curve['candle_begin_time'].iloc[0]) * 365) - 1
results.loc[0, '年化收益'] = str(round(annual_return, 2)) + ' 倍'

# ===计算最大回撤，最大回撤的含义：《如何通过3行代码计算最大回撤》https://mp.weixin.qq.com/s/Dwt4lkKR_PEnWRprLlvPVw
# 计算当日之前的资金曲线的最高点
equity_curve['max2here'] = equity_curve['equity_curve'].expanding().max()
# 计算到历史最高值到当日的跌幅，drowdwon
equity_curve['dd2here'] = equity_curve['equity_curve'] / equity_curve['max2here'] - 1
# 计算最大回撤，以及最大回撤结束时间
end_date, max_draw_down = tuple(equity_curve.sort_values(by=['dd2here']).iloc[0][['candle_begin_time', 'dd2here']])
# 计算最大回撤开始时间
start_date = equity_curve[equity_curve['candle_begin_time'] <= end_date].sort_values(by='equity_curve', ascending=False).iloc[0]['candle_begin_time']
# 将无关的变量删除
equity_curve.drop(['max2here', 'dd2here'], axis=1, inplace=True)
results.loc[0, '最大回撤'] = format(max_draw_down, '.2%')
results.loc[0, '最大回撤开始时间'] = str(start_date)
results.loc[0, '最大回撤结束时间'] = str(end_date)

# ===年化收益/回撤比
results.loc[0, '年化收益/回撤比'] = round(abs(annual_return / max_draw_down), 2)

# ===统计每笔交易
results.loc[0, '盈利笔数'] = len(trade.loc[trade['change'] > 0])  # 盈利笔数
results.loc[0, '亏损笔数'] = len(trade.loc[trade['change'] <= 0])  # 亏损笔数
results.loc[0, '胜率'] = format(results.loc[0, '盈利笔数'] / len(trade), '.2%')  # 胜率

results.loc[0, '每笔交易平均盈亏'] = format(trade['change'].mean(), '.2%')  # 每笔交易平均盈亏
results.loc[0, '盈亏收益比'] = round(trade.loc[trade['change'] > 0]['change'].mean() / \
                                trade.loc[trade['change'] < 0]['change'].mean() * (-1), 2)  # 盈亏比

results.loc[0, '单笔最大盈利'] = format(trade['change'].max(), '.2%')  # 单笔最大盈利
results.loc[0, '单笔最大亏损'] = format(trade['change'].min(), '.2%')  # 单笔最大亏损

# ===统计持仓时间，会比实际时间少一根K线的是距离
trade['持仓时间'] = trade['end_bar'] - trade.index
max_days, max_seconds = trade['持仓时间'].max().days, trade['持仓时间'].max().seconds
max_hours = max_seconds // 3600
max_minute = (max_seconds - max_hours * 3600) // 60
results.loc[0, '单笔最长持有时间'] = str(max_days) + ' 天 ' + str(max_hours) + ' 小时 ' + str(max_minute) + ' 分钟'  # 单笔最长持有时间

min_days, min_seconds = trade['持仓时间'].min().days, trade['持仓时间'].min().seconds
min_hours = min_seconds // 3600
min_minute = (min_seconds - min_hours * 3600) // 60
results.loc[0, '单笔最短持有时间'] = str(min_days) + ' 天 ' + str(min_hours) + ' 小时 ' + str(min_minute) + ' 分钟'  # 单笔最短持有时间

mean_days, mean_seconds = trade['持仓时间'].mean().days, trade['持仓时间'].mean().seconds
mean_hours = mean_seconds // 3600
mean_minute = (mean_seconds - mean_hours * 3600) // 60
results.loc[0, '平均持仓周期'] = str(mean_days) + ' 天 ' + str(mean_hours) + ' 小时 ' + str(mean_minute) + ' 分钟'  # 平均持仓周期

# ===连续盈利亏算
results.loc[0, '最大连续盈利笔数'] = max(
    [len(list(v)) for k, v in itertools.groupby(np.where(trade['change'] > 0, 1, np.nan))])  # 最大连续盈利笔数
results.loc[0, '最大连续亏损笔数'] = max(
    [len(list(v)) for k, v in itertools.groupby(np.where(trade['change'] < 0, 1, np.nan))])  # 最大连续亏损笔数

# ===每月收益率
equity_curve.set_index('candle_begin_time', inplace=True)
monthly_return = equity_curve[['equity_change']].resample(rule='M').apply(lambda x: (1 + x).prod() - 1)

return results.T, monthly_return

**《python数字货币量化交易》学习笔记  第四部分别有洞天篇   4.1 择时策略实盘与优化 4.1.2 资金曲线图 Statistics**

"""
# 课程内容
策略评价函数
"""
import pandas as pd
import numpy as np
import itertools


# ======= 策略评价 =========
# 将资金曲线数据，转化为交易数据
def transfer_equity_curve_to_trade(equity_curve):
    """
    将资金曲线数据，转化为一笔一笔的交易
    :param equity_curve: 资金曲线函数计算好的结果，必须包含pos
    :return:
    """
    # =选取开仓、平仓条件
    condition1 = equity_curve['pos'] != 0
    condition2 = equity_curve['pos'] != equity_curve['pos'].shift(1)
    open_pos_condition = condition1 & condition2

    # =计算每笔交易的start_time
    if 'start_time' not in equity_curve.columns:
        equity_curve.loc[open_pos_condition, 'start_time'] = equity_curve['candle_begin_time']
        equity_curve['start_time'].fillna(method='ffill', inplace=True)
        equity_curve.loc[equity_curve['pos'] == 0, 'start_time'] = pd.NaT

    # =对每次交易进行分组，遍历每笔交易
    trade = pd.DataFrame()  # 计算结果放在trade变量中

    for _index, group in equity_curve.groupby('start_time'):

        # 记录每笔交易
        # 本次交易方向
        trade.loc[_index, 'signal'] = group['pos'].iloc[0]

        # 本次交易杠杆倍数
        if 'leverage_rate' in group:
            trade.loc[_index, 'leverage_rate'] = group['leverage_rate'].iloc[0]

        g = group[group['pos'] != 0]  # 去除pos=0的行
        # 本次交易结束那根K线的开始时间
        trade.loc[_index, 'end_bar'] = g.iloc[-1]['candle_begin_time']
        # 开仓价格
        trade.loc[_index, 'start_price'] = g.iloc[0]['open']
        # 平仓信号的价格
        trade.loc[_index, 'end_price'] = g.iloc[-1]['close']
        # 持仓k线数量
        trade.loc[_index, 'bar_num'] = g.shape[0]
        # 本次交易收益
        trade.loc[_index, 'change'] = (group['equity_change'] + 1).prod() - 1
        # 本次交易结束时资金曲线
        trade.loc[_index, 'end_equity_curve'] = g.iloc[-1]['equity_curve']
        # 本次交易中资金曲线最低值
        trade.loc[_index, 'min_equity_curve'] = g['equity_curve'].min()

    return trade


# 计算策略评价指标
def strategy_evaluate(equity_curve, trade):
    """
    :param equity_curve: 带资金曲线的df
    :param trade: transfer_equity_curve_to_trade的输出结果，每笔交易的df
    :return:
    """

    # ===新建一个dataframe保存回测指标
    results = pd.DataFrame()

    # ===计算累积净值
    results.loc[0, '累积净值'] = round(equity_curve['equity_curve'].iloc[-1], 2)

    # ===计算年化收益
    annual_return = (equity_curve['equity_curve'].iloc[-1] / equity_curve['equity_curve'].iloc[0]) ** (
        '1 days 00:00:00' / (equity_curve['candle_begin_time'].iloc[-1] - equity_curve['candle_begin_time'].iloc[0]) * 365) - 1
    results.loc[0, '年化收益'] = str(round(annual_return, 2)) + ' 倍'

    # ===计算最大回撤，最大回撤的含义：《如何通过3行代码计算最大回撤》https://mp.weixin.qq.com/s/Dwt4lkKR_PEnWRprLlvPVw
    # 计算当日之前的资金曲线的最高点
    equity_curve['max2here'] = equity_curve['equity_curve'].expanding().max()
    # 计算到历史最高值到当日的跌幅，drowdwon
    equity_curve['dd2here'] = equity_curve['equity_curve'] / equity_curve['max2here'] - 1
    # 计算最大回撤，以及最大回撤结束时间
    end_date, max_draw_down = tuple(equity_curve.sort_values(by=['dd2here']).iloc[0][['candle_begin_time', 'dd2here']])
    # 计算最大回撤开始时间
    start_date = equity_curve[equity_curve['candle_begin_time'] <= end_date].sort_values(by='equity_curve', ascending=False).iloc[0]['candle_begin_time']
    # 将无关的变量删除
    equity_curve.drop(['max2here', 'dd2here'], axis=1, inplace=True)
    results.loc[0, '最大回撤'] = format(max_draw_down, '.2%')
    results.loc[0, '最大回撤开始时间'] = str(start_date)
    results.loc[0, '最大回撤结束时间'] = str(end_date)

    # ===年化收益/回撤比
    results.loc[0, '年化收益/回撤比'] = round(abs(annual_return / max_draw_down), 2)

    # ===统计每笔交易
    results.loc[0, '盈利笔数'] = len(trade.loc[trade['change'] > 0])  # 盈利笔数
    results.loc[0, '亏损笔数'] = len(trade.loc[trade['change'] <= 0])  # 亏损笔数
    results.loc[0, '胜率'] = format(results.loc[0, '盈利笔数'] / len(trade), '.2%')  # 胜率

    results.loc[0, '每笔交易平均盈亏'] = format(trade['change'].mean(), '.2%')  # 每笔交易平均盈亏
    results.loc[0, '盈亏收益比'] = round(trade.loc[trade['change'] > 0]['change'].mean() / \
                                    trade.loc[trade['change'] < 0]['change'].mean() * (-1), 2)  # 盈亏比

    results.loc[0, '单笔最大盈利'] = format(trade['change'].max(), '.2%')  # 单笔最大盈利
    results.loc[0, '单笔最大亏损'] = format(trade['change'].min(), '.2%')  # 单笔最大亏损

    # ===统计持仓时间，会比实际时间少一根K线的是距离
    trade['持仓时间'] = trade['end_bar'] - trade.index
    max_days, max_seconds = trade['持仓时间'].max().days, trade['持仓时间'].max().seconds
    max_hours = max_seconds // 3600
    max_minute = (max_seconds - max_hours * 3600) // 60
    results.loc[0, '单笔最长持有时间'] = str(max_days) + ' 天 ' + str(max_hours) + ' 小时 ' + str(max_minute) + ' 分钟'  # 单笔最长持有时间

    min_days, min_seconds = trade['持仓时间'].min().days, trade['持仓时间'].min().seconds
    min_hours = min_seconds // 3600
    min_minute = (min_seconds - min_hours * 3600) // 60
    results.loc[0, '单笔最短持有时间'] = str(min_days) + ' 天 ' + str(min_hours) + ' 小时 ' + str(min_minute) + ' 分钟'  # 单笔最短持有时间

    mean_days, mean_seconds = trade['持仓时间'].mean().days, trade['持仓时间'].mean().seconds
    mean_hours = mean_seconds // 3600
    mean_minute = (mean_seconds - mean_hours * 3600) // 60
    results.loc[0, '平均持仓周期'] = str(mean_days) + ' 天 ' + str(mean_hours) + ' 小时 ' + str(mean_minute) + ' 分钟'  # 平均持仓周期

    # ===连续盈利亏算
    results.loc[0, '最大连续盈利笔数'] = max(
        [len(list(v)) for k, v in itertools.groupby(np.where(trade['change'] > 0, 1, np.nan))])  # 最大连续盈利笔数
    results.loc[0, '最大连续亏损笔数'] = max(
        [len(list(v)) for k, v in itertools.groupby(np.where(trade['change'] < 0, 1, np.nan))])  # 最大连续亏损笔数

    # ===每月收益率
    equity_curve.set_index('candle_begin_time', inplace=True)
    monthly_return = equity_curve[['equity_change']].resample(rule='M').apply(lambda x: (1 + x).prod() - 1)

    return results.T, monthly_return