from fastapi import FastAPI, Request, Response # from fastapi.responses import ORJSONResponse from pydantic import BaseModel from pydantic import TypeAdapter from utils.prepare import get_dataframe, prepare_data, get_forecast_data from utils.process import find_best_model_by_metric from utils.predict import forecast from typing import List import numpy as np import pandas as pd from pandas.tseries.offsets import CustomBusinessDay from pandas.tseries.holiday import USFederalHolidayCalendar from datetime import datetime import os import threading import requests from contextlib import asynccontextmanager from fastapi import FastAPI # import telebot # from telebot import types import uvicorn import asyncio # from vercel_bot_v2.stock.utils.process import find_best_model_by_metric # CRITICAL: threaded=False is required for serverless (Vercel)! TELEGRAM_TOKEN = os.environ.get("TELEGRAM_TOKEN_V2") WEBHOOK_URL = os.environ.get( "WEBHOOK_URL_V2" ) # e.g., https://your-project.vercel.app/webhook # bot = telebot.TeleBot(TELEGRAM_TOKEN, threaded=False) # Replace this with your actual helper function if it's in another file def convert_to_serializable(data): return str(data) # # --- Telegram Bot Handlers --- # @bot.message_handler(commands=["start", "help"]) # def send_welcome(message): # bot.reply_to( # message, "Hi! Send me a stock ticker (e.g., AAPL) and I will predict it." # ) def parse_input(message): parts = message.split(" ") # Mandatory ticker if not parts: raise ValueError("Ticker is mandatory") ticker = parts[0].strip().upper() # Optional arguments with defaults metric = "mae_bins" # default leverage = 1.0 # default (float) include_short = False # default (bool) # Parse remaining parts if they exist if len(parts) > 1: metric = parts[1] # keep as uppercase string, or .lower() if needed if len(parts) > 2: try: leverage = float(parts[2]) except ValueError: raise ValueError(f"Leverage must be a number, got '{parts[2]}'") if len(parts) > 3: val = parts[3] if val in ("TRUE", "1", "YES", "Y"): include_short = True elif val in ("FALSE", "0", "NO", "N"): include_short = False else: raise ValueError(f"include_short must be TRUE/FALSE or 1/0, got '{val}'") return ticker, metric, leverage, include_short # @bot.message_handler( # func=lambda message: not message.text.startswith("/"), content_types=["text"] # ) # def handle_prediction(message): # # user_input = message.text.strip().split("\n") # # processing_msg = bot.reply_to(message, f"Fetching prediction for...") # # try: # # ticker = user_input[0].strip().upper() # # q = int(user_input[1].strip()) # # start = user_input[2].strip() # # if len(user_input) > 3: # # end = user_input[3].strip() # # else: # # end = "2030-10-10" # # if len(user_input) > 4: # # fee = float(user_input[4].strip()) # # else: # # fee = 0.003 # # except Exception as e: # # bot.edit_message_text( # # chat_id=processing_msg.chat.id, # # message_id=processing_msg.message_id, # # text=f"An error occurred: {str(e)}", # # ) # # return # ticker, metric, leverage, include_short = parse_input(message.text.strip()) # if "acc" in metric: # is_smaller = False # else: # is_smaller = True # # Show "typing..." status in Telegram while the API is working # bot.send_chat_action(message.chat.id, "typing") # forecast_res_clean = "Failed to retrieve data." # ticker = message.text.strip().upper() # processing_msg = bot.reply_to(message, f"Fetching prediction for {ticker}...") # # print("Processing Telegram....") # bot.edit_message_text( # chat_id=processing_msg.chat.id, # message_id=processing_msg.message_id, # text="otw processing message", # ) # try: # # headers = { # # "Authorization": f"Bearer {AUTH_KEY}", # # "Content-Type": "application/json" # # } # # payload = {"ticker": ticker} # Change 'ticker' to 'symbol' if the API requires it # # response = requests.post(API_URL, json=payload, headers=headers) # fee = 0.003 # # include_short = False # # ticker = ticker.upper() # start = "2020-01-01" # # end = "2030-01-01" # q = 3 # # conditions = [x.model_dump() for x in req.conditions] # # conditions = [{"operator": "<", "logic": "&", "value": q - 1}] # train_df, test_df, feature_cols = get_dataframe( # ticker, q, start=start, end=None # ) # train_results = prepare_data(train_df, test_df, feature_cols) # best_seq_len, best_value, selected_model = find_best_model_by_metric( # train_results, metric, is_smaller=is_smaller # ) # bot.edit_message_text( # chat_id=processing_msg.chat.id, # message_id=processing_msg.message_id, # text="Sudah mendapatkan Markov_Chain", # ) # results = forecast( # test_df, # q, # best_seq_len, # selected_model, # fee=0.003, # leverage=leverage, # include_short=False, # ) # # _, train_preds = create_majority_pred(train, transition) # # _, test_preds = create_majority_pred(test, transition) # # eval_res = evaluate_markov_chain(train, test, transition, q) # # test_label = get_accuracy_detail(test, test_preds) # # train_label = get_accuracy_detail(train, train_preds) # # test_label_acc = test_label.set_index("label").to_dict() # # train_label_acc = train_label.set_index("label").to_dict() # # bot.edit_message_text( # # chat_id=processing_msg.chat.id, # # message_id=processing_msg.message_id, # # text="Tinggal forecast saja", # # ) # # forecast_res = forecast( # # test["ret"], test_preds, conditions, fee=fee, include_short=include_short # # ) # # df_cast = get_forecast_data(ticker, q, start=start, end=end) # # data_cast = one_day_future(df_cast, conditions, transition) # ( # pred_bins_tomorrow, # trend_signal_for_tomorrow, # signal_tomorrow, # current_date, # ) = get_forecast_data( # ticker, q, feature_cols, best_seq_len, selected_model, start=start # ) # us_cal = CustomBusinessDay(calendar=USFederalHolidayCalendar()) # forecast_date = current_date + us_cal # results.update( # { # "selected_seq": best_seq_len, # } # ) # bot.edit_message_text( # chat_id=processing_msg.chat.id, # message_id=processing_msg.message_id, # text="Sudah dapat forecast_res_clean", # ) # # bot.edit_message_text( # # chat_id=processing_msg.chat.id, # # message_id=processing_msg.message_id, # # text=f"{forecast_res_clean}", # # ) # # return forecast_res # # return ORJSONResponse(content=TypeAdapter(dict).dump_python(forecast_res, mode="json")) # # forecast_res_clean = convert_to_serializable( # # forecast_res # # ) # using the helper above # # text = f"Current Date: {forecast_res['forecast']['current_date']}\nForecast Date: {forecast_res['forecast']['forecast_date']}\nReturn vs Price difference: {forecast_res['return_difference']:.3f}\n\nSIGNAL: {forecast_res['forecast']['signal']}" # # text += f"\n\nBase Sharpe: {forecast_res['base_sharpe']:.3f}\nBase MaxDD: {forecast_res['base_maxdd']}\nBase CAGR: {forecast_res['base_cagr']:.3f}\nBase Return: {forecast_res['base_return']:.3f}\n\nSharpe: {forecast_res['sharpe']:.3f}\nMaxDD: {forecast_res['maxdd']:.3f}\nCAGR: {forecast_res['cagr']:.3f}\nFinal Return: {forecast_res['final_return']:.3f}\n\nWinrate: {forecast_res['winrate']*100:.3f}%" # # # text += f"\n" # # text += f"\n\nNum Buy/Hold Signals: {forecast_res['num_long_signals']}\nNum Skip/Exit Signals: {forecast_res['num_exit_signals']}\nNum Change Positions: {forecast_res['num_change_positions']}\nNum Trades: {forecast_res['num_trades']}\nTotal Testing Days: {forecast_res['total_days']}" # output = ( # f"Current Date: {current_date}\n" # f"Forecast Date: {forecast_date}\n\n" # f"Signal Tomorrow: {signal_tomorrow}\n\n" # f"Pred Bins Tomorrow: {pred_bins_tomorrow}\n\n" # f"Trend Signal Tomorrow: {trend_signal_for_tomorrow}\n\n" # ) # # Loop through each strategy in the results dictionary # for strategy_name, metrics in results.items(): # if strategy_name in ["leverage", "fee", "chart_path"]: # continue # output += f"{strategy_name}\n" # output += f" Final Equity: {metrics['Final Equity']:.4f}\n" # output += f" Sharpe: {metrics['Sharpe']:.3f}\n" # output += f" CAGR: {metrics['CAGR']:.3f}\n" # output += f" MaxDD: {metrics['MaxDD']:.3f}\n" # output += f" VaR 95%: {metrics['VaR 95%']:.3f}\n" # output += f" Kelly: {metrics['Kelly']:.3f}\n\n" # for x in ["leverage", "fee"]: # output += f"{x}: {results[x]}\n" # bot.edit_message_text( # chat_id=processing_msg.chat.id, # message_id=processing_msg.message_id, # # text=f"{forecast_res_clean}", # text=output, # ) # except Exception as e: # bot.edit_message_text( # chat_id=processing_msg.chat.id, # message_id=processing_msg.message_id, # text=f"An error occurred: {str(e)}", # ) # return # # except Exception as e: # # forecast_res_clean = f"An error occurred: {str(e)}" # # Reply directly (editing messages is complex in serverless webhooks) # # bot.reply_to(message, f"Prediction for {ticker}:\n\n{str(forecast_res_clean)}") # chart_path = results.get("chart_path") # if chart_path and os.path.exists(chart_path): # try: # with open(chart_path, "rb") as photo_file: # bot.send_photo( # chat_id=message.chat.id, # photo=photo_file, # caption=f"📈 Chart forecast for {ticker}", # ) # except Exception as e: # bot.reply_to(message, f"Failed to send chart: {str(e)}") # # 3. CLEANUP: Delete the file from /tmp/ to prevent memory bloat # os.remove(chart_path) # --- FastAPI Setup --- app = FastAPI() # @app.post("/webhook") # async def telegram_webhook(request: Request): # # 1. Parse the incoming update from Telegram # request_body_dict = await request.json() # update = types.Update.de_json(request_body_dict) # # 2. Process the update. # # We use asyncio.to_thread so the sync telebot code doesn't block FastAPI's event loop. # await asyncio.to_thread(bot.process_new_updates, [update]) # # 3. Return 200 OK immediately so Telegram knows we received it # return Response(status_code=200) # @app.get("/set_webhook") # async def set_webhook(): # """ # Call this endpoint ONCE in your browser after deploying to register your URL with Telegram. # """ # if not WEBHOOK_URL: # return {"error": "WEBHOOK_URL environment variable is not set."} # bot.remove_webhook() # bot.set_webhook(url=WEBHOOK_URL) # return {"message": f"Webhook successfully set to {WEBHOOK_URL}"} class Cond(BaseModel): logic: str | None = None operator: str = "<" value: int = 6 class Item(BaseModel): ticker: str = "AAPL" start: str = "2017-01-01" end: str = "2026-06-01" q: int = 5 fee: float | None = 0.3 include_short: bool | None = False conditions: List[Cond] @app.get("/") def greet_json(): return {"Hello": "World!"} def convert_to_serializable(obj): """Recursively convert NumPy types, pandas Timestamp, and datetime to JSON-serializable types.""" if isinstance(obj, dict): return {k: convert_to_serializable(v) for k, v in obj.items()} elif isinstance(obj, (list, tuple)): return [convert_to_serializable(v) for v in obj] elif isinstance(obj, (np.integer, np.int64)): return int(obj) elif isinstance(obj, (np.floating, np.float64)): return float(obj) elif isinstance(obj, np.ndarray): return obj.tolist() elif isinstance(obj, np.bool_): return bool(obj) elif isinstance(obj, pd.Timestamp): # Convert to ISO 8601 string (e.g., "2024-01-01T00:00:00") return obj.isoformat() elif isinstance(obj, datetime): return obj.isoformat() elif isinstance(obj, pd.Period): # Convert period to string (e.g., "2024-01") return str(obj) elif isinstance(obj, (pd.Series, pd.DataFrame)): # Better to avoid passing whole DataFrames, but if needed: return convert_to_serializable(obj.to_dict()) else: return obj class ItemV(BaseModel): ticker: str = "AAPL" metric: str = "mae_bins" leverage: float = 1.0 include_short: bool = False @app.post("/predict") def get_predict(req: ItemV): ticker, metric, leverage, include_short = ( req.ticker, req.metric, req.leverage, req.include_short, ) if "acc" in metric: is_smaller = False else: is_smaller = True try: fee = 0.003 # include_short = False # ticker = ticker.upper() start = "2020-01-01" # end = "2030-01-01" q = 3 train_df, test_df, feature_cols = get_dataframe( ticker, q, start=start, end=None ) train_results = prepare_data(train_df, test_df, feature_cols) best_seq_len, best_value, selected_model = find_best_model_by_metric( train_results, metric, is_smaller=is_smaller ) results = forecast( test_df, q, best_seq_len, selected_model, fee=fee, leverage=leverage, include_short=include_short, ) ( pred_bins_tomorrow, trend_signal_for_tomorrow, signal_tomorrow, current_date, ) = get_forecast_data( ticker, q, feature_cols, best_seq_len, selected_model, start=start ) us_cal = CustomBusinessDay(calendar=USFederalHolidayCalendar()) forecast_date = current_date + us_cal # results.update( # { # "selected_seq": best_seq_len, # } # ) output = ( f"Current Date: {current_date}\n" f"Forecast Date: {forecast_date}\n\n" f"Signal Tomorrow: {signal_tomorrow}\n\n" f"Pred Bins Tomorrow: {pred_bins_tomorrow}\n\n" f"Trend Signal Tomorrow: {trend_signal_for_tomorrow}\n\n" ) # Loop through each strategy in the results dictionary for strategy_name, metrics in results.items(): if strategy_name in ["leverage", "fee", "chart_path"]: continue if not isinstance(metrics, dict): continue output += f"{strategy_name}\n" output += f" Final Equity: {metrics['Final Equity']:.4f}\n" output += f" Sharpe: {metrics['Sharpe']:.3f}\n" output += f" CAGR: {metrics['CAGR']:.3f}\n" output += f" MaxDD: {metrics['MaxDD']:.3f}\n" output += f" VaR 95%: {metrics['VaR 95%']:.3f}\n" output += f" Kelly: {metrics['Kelly']:.3f}\n\n" for x in ["leverage", "fee"]: output += f"{x}: {results[x]}\n" output += "\n" for x in ["acc", "mae_bins", "adj_acc", "val_loss"]: output += f"{x}: {selected_model[x]:.3f}\n" output += f"Selected Length: {best_seq_len}" except Exception as e: raise e # except Exception as e: # forecast_res_clean = f"An error occurred: {str(e)}" # Reply directly (editing messages is complex in serverless webhooks) # bot.reply_to(message, f"Prediction for {ticker}:\n\n{str(forecast_res_clean)}") chart_path = results.get("chart_path") if chart_path and os.path.exists(chart_path): # try: # with open(chart_path, "rb") as photo_file: # bot.send_photo( # chat_id=message.chat.id, # photo=photo_file, # caption=f"📈 Chart forecast for {ticker}", # ) # except Exception as e: # bot.reply_to(message, f"Failed to send chart: {str(e)}") # 3. CLEANUP: Delete the file from /tmp/ to prevent memory bloat os.remove(chart_path) return {"text": output} if __name__ == "__main__": # uvicorn.run(app, host="0.0.0.0", port=8000, reload=True) uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=True) # @app.post("/predict") # def get_predict(req: Item): # fee = req.fee # include_short = req.include_short # ticker = req.ticker.upper() # start = req.start # end = req.end # q = req.q # conditions = [x.model_dump() for x in req.conditions] # df = get_dataframe(ticker, q, start=start, end=end) # train, test, transition = get_markov_chain(df) # _, train_preds = create_majority_pred(train, transition) # _, test_preds = create_majority_pred(test, transition) # eval_res = evaluate_markov_chain(train, test, transition, q) # test_label = get_accuracy_detail(test, test_preds) # train_label = get_accuracy_detail(train, train_preds) # test_label_acc = test_label.set_index("label").to_dict() # train_label_acc = train_label.set_index("label").to_dict() # forecast_res = forecast( # test["ret"], test_preds, conditions, fee=fee, include_short=include_short # ) # df_cast = get_forecast_data(ticker, q, start=start, end=end) # data_cast = one_day_future(df_cast, conditions, transition) # forecast_res.update( # { # "ticker": ticker, # "eval_res": eval_res, # "test_label_accuracy_details": test_label_acc, # "train_label_accuracy_details": train_label_acc, # "forecast": data_cast, # } # ) # # return forecast_res # # return ORJSONResponse(content=TypeAdapter(dict).dump_python(forecast_res, mode="json")) # forecast_res_clean = convert_to_serializable(forecast_res) # using the helper above # return ORJSONResponse(content=forecast_res_clean)