utils.py

import io
import os
import json
import requests
import pandas as pd
import mplfinance as mpf
import matplotlib.pyplot as plt
from PIL import Image


def generate_pil_img():
    img_buf = io.BytesIO()
    plt.savefig(img_buf, format='png')
    return Image.open(img_buf)


def plot_Klines_fig(bars, code, precision=0.0001):
    fig, axes = mpf.plot(bars, type='candle', mav=(5, 10), volume=True, title=f'\n{code}',
                         figsize=(12, 6), returnfig=True)
    fixed_txt = 'Predict\nNext Bar'
    for ci in ['Open', 'High', 'Low', 'Close', 'Volume']:
        fixed_txt += f'\n{ci[0]}:{"{:.3e}".format(bars.iloc[-1][ci])}'
    hl_chg = bars['High'].iloc[-1] / bars['Low'].iloc[-1] - 1.0
    axes[0].annotate(fixed_txt, textcoords='axes fraction', xytext=(1.006, -0.37), color='r',
                     bbox=dict(facecolor='gray'),
                     xy=(len(bars) - 1, max(bars['Low'].min(), bars.iloc[-1]['Low'] * (1 - hl_chg / 4))),
                     arrowprops=dict(facecolor='r', width=5))
    return generate_pil_img()


def plot_txt(text):
    plt.figure(figsize=(12, 6))
    plt.text(0.2, 0.8, text, size=12, alpha=0.8)
    plt.axis('off')
    return generate_pil_img()


def Kline_predict_plot(model_id, symbols, user_opinion, frequency, feature_saturation, sensitivity,
                       latest_bar_completed, show_bars, once_max):
    params = {
        'codes_str': symbols,
        'user_opinion': user_opinion,
        'frequency': frequency,
        'feature_saturation': feature_saturation,
        'latest_completed': latest_bar_completed,
        'sensitivity': sensitivity,
        'show_bars': show_bars,
        'once_max': once_max,
    }
    headers = {'Content-Type': 'application/json'}
    response = requests.post(os.environ[model_id] + 'predict', data=json.dumps(params), headers=headers)
    result = response.json()
    coins_not_available = result['codes_remove']
    klines_data = result['klines_data']
    figs = []
    for codei, vi in klines_data.items():
        bars = pd.DataFrame(vi)
        bars.index = bars.index.map(pd.Timestamp)
        figs.append(plot_Klines_fig(bars, codei))
    if len(figs) < 1:
        figs = [plot_txt('No available symbols in `Symbols` box.')]
    return ','.join(coins_not_available) if len(coins_not_available) > 0 else 'No one', figs


def back_test_plot(model_id, symbols, frequency, feature_saturation, sensitivity, test_bars):
    params = {
        'codes_str': symbols,
        'frequency': frequency,
        'feature_saturation': feature_saturation,
        'sensitivity': sensitivity,
        'test_bars': test_bars,
    }
    headers = {'Content-Type': 'application/json'}
    response = requests.post(os.environ[model_id] + 'back_test', data=json.dumps(params), headers=headers)
    result = response.json()
    test_result = result['test_result']
    if isinstance(test_result, str):
        return [plot_txt(test_result)]
    else:
        assert isinstance(test_result, dict), "`test_result` type is wrong."
        keys_list = list(test_result.keys())
        code, code_name = keys_list if len(keys_list[0]) < len(keys_list[1]) else keys_list[::-1]
        labels, profits = pd.Series(test_result[code]), pd.Series(test_result[code_name])
        labels.index, profits.index = labels.index.map(pd.Timestamp), profits.index.map(pd.Timestamp)
        plt.figure()
        (profits + 1.0).cumprod().plot(
            label=f'profits => win_ratio:{round((profits > 0).mean() * 100, 2)}%')
        (labels + 1.0).cumprod().plot(
            label=f'{code} => up_ratio:{round((labels > 0).mean() * 100, 2)}%', linestyle='--')
        plt.title(code_name)
        plt.legend()
        plt.grid()
        return [generate_pil_img()]