Spaces:
Runtime error
Runtime error
| 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()] | |