mt4report_enhanced.py

"""
Install requirements
> pip install pandas openpyxl python-dateutil more-itertools pymt5adapter pytz
"""

import argparse
import datetime as dt
import math
from pathlib import Path

import more_itertools as moit
import numpy as np
import pandas as pd
import pytz

import pymt5adapter as mta
from pymt5adapter.symbol import Symbol


def iter_datetimes(from_, to_, **td_kwargs):
    delta = dt.timedelta(**td_kwargs)
    time = from_
    while True:
        yield time
        if time >= to_:
            break
        time = min(time + delta, to_)


def get_mt5_ticks(symbol, from_, to_, flags):
    times = iter_datetimes(from_, to_, days=5)
    rates = [mta.copy_ticks_range(symbol, f, t, flags)
             for f, t in moit.pairwise(times)]
    stacked = np.hstack(rates)
    return stacked


def report(report_path, timezone, is_json=False, is_excel=True, is_csv=False):
    time_format = '%Y.%m.%d %H:%M'
    report_path = Path(report_path)
    print('Parsing MT4 report...')
    info, origin_df = pd.read_html(str(report_path), header=0, index_col=0, parse_dates=True)
    symbol = Symbol(symbol=info.columns[1][:6])
    refactored_trade_stats = []
    for trade in origin_df.groupby('Order').groups.values():
        trade_df = origin_df.loc[trade]
        open = trade_df.iloc[0, :].to_dict()
        close = trade_df.iloc[1, :].to_dict()
        order_type = open['Type']
        if order_type not in ['buy', 'sell']:
            continue
        order_type = mta.ORDER_TYPE.BUY if order_type == 'buy' else mta.ORDER_TYPE.SELL
        bid_or_ask = 'ask' if order_type is mta.ORDER_TYPE.SELL else 'bid'
        combined = close.copy()
        open_time, close_time = open['Time'], close['Time']
        open_price = open['Price']
        combined['OpenTime'] = open_time
        combined['CloseTime'] = combined.pop('Time')
        combined['OpenPrice'] = open_price
        combined['ClosePrice'] = close_price = combined.pop('Price')
        combined['Type'] = open['Type']
        time_begin = dt.datetime.strptime(open_time, time_format).replace(tzinfo=timezone)
        time_end = dt.datetime.strptime(close_time, time_format).replace(tzinfo=timezone)
        tick_data = get_mt5_ticks(symbol, time_begin, time_end, flags=mta.COPY_TICKS_ALL)
        tick_df = pd.DataFrame(tick_data)
        print(f'processing order #{combined["Order"]} with {len(tick_data)} ticks')
        prices = tick_df[bid_or_ask]
        tick_min = prices.min()  # min(open_price, close_price, prices.min())
        tick_max = prices.max()  # max(open_price, close_price, prices.max())
        mae_price, mfe_price = tick_min, tick_max
        ticks_won = symbol.normalize_price((close_price - open_price) / symbol.trade_tick_size)
        if order_type is mta.ORDER_TYPE.SELL:
            mae_price, mfe_price = mfe_price, mae_price
            ticks_won = -ticks_won

        def calc_profit(price):
            return mta.order_calc_profit(
                order_type=order_type,
                symbol=symbol,
                volume=float(combined['Size']),
                price_open=float(combined['OpenPrice']),
                price_close=price,
            )

        def calc_ticks(price):
            if order_type is mta.ORDER_TYPE.BUY:
                delta = price - open_price
            else:
                delta = open_price - price
            return symbol.normalize_price(delta / symbol.trade_tick_size)

        profit = float(combined['Profit'])
        prices = [mae_price, mfe_price]
        if any(math.isnan(p) for p in prices):
            continue
        mae_profit, mfe_profit = map(calc_profit, prices)
        mae_ticks, mfe_ticks = map(calc_ticks, prices)
        tick_range = abs(tick_max - tick_min)
        if order_type is mta.ORDER_TYPE.BUY:
            entry_efficiency = abs(tick_max - open_price) / tick_range
            exit_efficiency = abs(close_price - tick_min) / tick_range
        else:
            entry_efficiency = abs(open_price - tick_min) / tick_range
            exit_efficiency = abs(tick_max - close_price) / tick_range

        combined.update({
            'TotalTickCount' : len(tick_data),
            'TicksWon'       : ticks_won,
            'MAEprice'       : mae_price,
            'MFEprice'       : mfe_price,
            'MAEprofit'      : min(mae_profit, profit),
            'MFEprofit'      : max(mfe_profit, profit),
            'MAEticks'       : mae_ticks,
            'MFEticks'       : mfe_ticks,
            'EntryEfficiency': entry_efficiency,
            'ExitEfficiency' : exit_efficiency,
        })
        refactored_trade_stats.append(combined)

    result_df = pd.DataFrame(refactored_trade_stats)
    path_template = f'{report_path.parent}\\{report_path.stem}.{{}}'
    for _ in range(2):
        try:
            if is_excel:
                print('Saving to Excel...')
                result_df.to_excel(path_template.format('xlsx'), index=False)
            if is_csv:
                print('Saving to CSV...')
                result_df.to_csv(path_template.format('csv'))
            if is_json:
                print('Saving to JSON...')
                result_df.to_json(path_template.format('json'), indent=4)
        except PermissionError:
            input("Couldn't write to report file! If you have it open, close it and press Enter > ")
            continue
        break


def main():
    parser = argparse.ArgumentParser(
        description=('Calculate the MAE-MFE for a MT4 trade report and export to excel. '
                     'The excel file is saved to the root of the same directory of the report.')
    )
    parser.add_argument('path', help='Path to the strategy report')
    parser.add_argument('--timezone', '-tz', default='UTC', help='The timezone for the MT4 broker.')
    # parser.add_argument(
    #     '--offset', '-o', default=0, type=int,
    #     help='The number of hours offset from the Brokers (MT5time - MT4time) (default is 0)')
    parser.add_argument('--json', action='store_true', help='Export report to JSON.')
    parser.add_argument('--excel', action='store_true', help='Export report to Excel.')
    parser.add_argument('--csv', action='store_true', help='Export report to CSV.')
    args = parser.parse_args()
    timezone = pytz.timezone(args.timezone)
    report(
        report_path=args.path, timezone=timezone,
        is_csv=args.csv, is_excel=args.excel, is_json=args.json)


if __name__ == '__main__':
    with mta.connected(raise_on_errors=True):
        main()