FinHight / app.py
Starfish55's picture
Upload 8 files
23c518c verified
import os
import json
import time
import random
from collections import defaultdict
from datetime import date, datetime, timedelta
import gradio as gr
import pandas as pd
import finnhub
from io import StringIO
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
import torch
from transformers import AutoTokenizer, AutoModelForCausalLM, BitsAndBytesConfig
from peft import PeftModel
# Suppress warnings
import warnings
warnings.filterwarnings('ignore', category=UserWarning)
warnings.filterwarnings('ignore', category=FutureWarning)
warnings.filterwarnings('ignore', category=DeprecationWarning)
# ---------- CẀU HÌNH ---------------------------------------------------------
# FinGPT Model Configuration
FINGPT_MODEL_NAME = "Starfish55/fingpt-complete"
FINGPT_BASE_MODEL = "microsoft/DialoGPT-medium" # Base model for FinGPT
MAX_LENGTH = 1024
TEMPERATURE = 0.7
# RapidAPI Configuration
RAPIDAPI_HOST = "alpha-vantage.p.rapidapi.com"
# Load Finnhub API keys from single secret (multiple keys separated by newlines)
FINNHUB_KEYS_RAW = os.getenv("FINNHUB_KEYS", "")
if FINNHUB_KEYS_RAW:
FINNHUB_KEYS = [key.strip() for key in FINNHUB_KEYS_RAW.split('\n') if key.strip()]
else:
FINNHUB_KEYS = []
# Load RapidAPI keys from single secret (multiple keys separated by newlines)
RAPIDAPI_KEYS_RAW = os.getenv("RAPIDAPI_KEYS", "")
if RAPIDAPI_KEYS_RAW:
RAPIDAPI_KEYS = [key.strip() for key in RAPIDAPI_KEYS_RAW.split('\n') if key.strip()]
else:
RAPIDAPI_KEYS = []
# Load Hugging Face API token for FinGPT model access
HF_TOKEN = os.getenv("HF_TOKEN", "")
if not HF_TOKEN:
print("⚠️ Warning: No Hugging Face token found in secrets")
# Filter out empty keys
FINNHUB_KEYS = [key for key in FINNHUB_KEYS if key.strip()]
# Validate that we have at least one key for each service
if not FINNHUB_KEYS:
print("⚠️ Warning: No Finnhub API keys found in secrets")
if not RAPIDAPI_KEYS:
print("⚠️ Warning: No RapidAPI keys found in secrets")
if not HF_TOKEN:
print("⚠️ Warning: No Hugging Face token found in secrets")
print("=" * 50)
print("πŸš€ FinRobot Forecaster Starting Up...")
print("=" * 50)
if FINNHUB_KEYS:
print(f"πŸ“Š Finnhub API: {len(FINNHUB_KEYS)} keys loaded")
else:
print("πŸ“Š Finnhub API: Not configured")
if RAPIDAPI_KEYS:
print(f"πŸ“ˆ RapidAPI Alpha Vantage: {RAPIDAPI_HOST} ({len(RAPIDAPI_KEYS)} keys loaded)")
else:
print("πŸ“ˆ RapidAPI Alpha Vantage: Not configured")
if HF_TOKEN:
print(f"πŸ€– FinGPT Model: {FINGPT_MODEL_NAME} loaded")
else:
print("πŸ€– FinGPT Model: Not configured")
print("βœ… Application started successfully!")
print("=" * 50)
# Initialize FinGPT model (if token available)
if HF_TOKEN:
try:
# Configure BitsAndBytesConfig for memory efficiency
bnb_config = BitsAndBytesConfig(
load_in_4bit=True,
bnb_4bit_use_double_quant=True,
bnb_4bit_quant_type="nf4",
bnb_4bit_compute_dtype=torch.bfloat16
)
# Load tokenizer and model
tokenizer = AutoTokenizer.from_pretrained(
FINGPT_MODEL_NAME,
token=HF_TOKEN,
trust_remote_code=True
)
model = AutoModelForCausalLM.from_pretrained(
FINGPT_MODEL_NAME,
token=HF_TOKEN,
quantization_config=bnb_config,
device_map="auto",
trust_remote_code=True
)
print(f"βœ… FinGPT model {FINGPT_MODEL_NAME} loaded successfully")
except Exception as e:
print(f"❌ Error loading FinGPT model: {e}")
model = None
tokenizer = None
else:
print("⚠️ FinGPT not configured - will use mock responses")
model = None
tokenizer = None
# CαΊ₯u hΓ¬nh Finnhub client (if keys available)
if FINNHUB_KEYS:
# Configure with first key for initial setup
finnhub_client = finnhub.Client(api_key=FINNHUB_KEYS[0])
print(f"βœ… Finnhub configured with {len(FINNHUB_KEYS)} keys")
else:
finnhub_client = None
print("⚠️ Finnhub not configured - will use mock news data")
# TαΊ‘o session vα»›i retry strategy cho requests
def create_session():
session = requests.Session()
retry_strategy = Retry(
total=3,
backoff_factor=1,
status_forcelist=[429, 500, 502, 503, 504],
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
return session
# TαΊ‘o session global
requests_session = create_session()
SYSTEM_PROMPT = (
"You are a seasoned stock-market analyst. "
"Given recent company news and optional basic financials, "
"return:\n"
"[Positive Developments] – 2-4 bullets\n"
"[Potential Concerns] – 2-4 bullets\n"
"[Prediction & Analysis] – a one-week price outlook with rationale."
)
# ---------- UTILITY HELPERS ----------------------------------------
def today() -> str:
return date.today().strftime("%Y-%m-%d")
def n_weeks_before(date_string: str, n: int) -> str:
return (datetime.strptime(date_string, "%Y-%m-%d") -
timedelta(days=7 * n)).strftime("%Y-%m-%d")
# ---------- DATA FETCHING --------------------------------------------------
def get_stock_data(symbol: str, steps: list[str]) -> pd.DataFrame:
# Thα»­ tαΊ₯t cαΊ£ RapidAPI Alpha Vantage keys
for rapidapi_key in RAPIDAPI_KEYS:
try:
print(f"πŸ“ˆ Fetching stock data for {symbol} via RapidAPI (key: {rapidapi_key[:8]}...)")
# RapidAPI Alpha Vantage endpoint
url = f"https://{RAPIDAPI_HOST}/query"
headers = {
"X-RapidAPI-Host": RAPIDAPI_HOST,
"X-RapidAPI-Key": rapidapi_key
}
params = {
"function": "TIME_SERIES_DAILY",
"symbol": symbol,
"outputsize": "full",
"datatype": "csv"
}
# Thα»­ lαΊ‘i 3 lαΊ§n vα»›i RapidAPI key hiện tαΊ‘i
for attempt in range(3):
try:
resp = requests_session.get(url, headers=headers, params=params, timeout=30)
if not resp.ok:
print(f"RapidAPI HTTP error {resp.status_code} with key {rapidapi_key[:8]}..., attempt {attempt + 1}")
time.sleep(2 ** attempt)
continue
text = resp.text.strip()
if text.startswith("{"):
info = resp.json()
msg = info.get("Note") or info.get("Error Message") or info.get("Information") or str(info)
if "rate limit" in msg.lower() or "quota" in msg.lower():
print(f"RapidAPI rate limit hit with key {rapidapi_key[:8]}..., trying next key")
break # Thα»­ key tiαΊΏp theo
raise RuntimeError(f"RapidAPI Alpha Vantage Error: {msg}")
# Parse CSV data
df = pd.read_csv(StringIO(text))
date_col = "timestamp" if "timestamp" in df.columns else df.columns[0]
df[date_col] = pd.to_datetime(df[date_col])
df = df.sort_values(date_col).set_index(date_col)
data = {"Start Date": [], "End Date": [], "Start Price": [], "End Price": []}
for i in range(len(steps) - 1):
s_date = pd.to_datetime(steps[i])
e_date = pd.to_datetime(steps[i+1])
seg = df.loc[s_date:e_date]
if seg.empty:
raise RuntimeError(
f"RapidAPI Alpha Vantage cannot get {symbol} data for {steps[i]} – {steps[i+1]}"
)
data["Start Date"].append(seg.index[0])
data["Start Price"].append(seg["close"].iloc[0])
data["End Date"].append(seg.index[-1])
data["End Price"].append(seg["close"].iloc[-1])
time.sleep(1) # RapidAPI has higher limits
print(f"βœ… Successfully retrieved {symbol} data via RapidAPI (key: {rapidapi_key[:8]}...)")
return pd.DataFrame(data)
except requests.exceptions.Timeout:
print(f"RapidAPI timeout with key {rapidapi_key[:8]}..., attempt {attempt + 1}")
if attempt < 2:
time.sleep(5 * (attempt + 1))
continue
else:
break
except requests.exceptions.RequestException as e:
print(f"RapidAPI request error with key {rapidapi_key[:8]}..., attempt {attempt + 1}: {e}")
if attempt < 2:
time.sleep(3)
continue
else:
break
except Exception as e:
print(f"RapidAPI Alpha Vantage failed with key {rapidapi_key[:8]}...: {e}")
continue # Thα»­ key tiαΊΏp theo
# Fallback: TαΊ‘o mock data nαΊΏu tαΊ₯t cαΊ£ RapidAPI keys đều fail
print("⚠️ All RapidAPI keys failed, using mock data for demonstration...")
return create_mock_stock_data(symbol, steps)
def create_mock_stock_data(symbol: str, steps: list[str]) -> pd.DataFrame:
"""TαΊ‘o mock data để demo khi API khΓ΄ng hoαΊ‘t Δ‘α»™ng"""
import numpy as np
data = {"Start Date": [], "End Date": [], "Start Price": [], "End Price": []}
# GiΓ‘ cΖ‘ bαΊ£n khΓ‘c nhau cho cΓ‘c symbol khΓ‘c nhau
base_prices = {
"AAPL": 180.0, "MSFT": 350.0, "GOOGL": 140.0,
"TSLA": 200.0, "NVDA": 450.0, "AMZN": 150.0
}
base_price = base_prices.get(symbol.upper(), 150.0)
for i in range(len(steps) - 1):
s_date = pd.to_datetime(steps[i])
e_date = pd.to_datetime(steps[i+1])
# TαΊ‘o giΓ‘ ngαΊ«u nhiΓͺn vα»›i xu hΖ°α»›ng tΔƒng nhαΊΉ
start_price = base_price + np.random.normal(0, 5)
end_price = start_price + np.random.normal(2, 8) # Xu hướng tăng nhẹ
data["Start Date"].append(s_date)
data["Start Price"].append(round(start_price, 2))
data["End Date"].append(e_date)
data["End Price"].append(round(end_price, 2))
base_price = end_price # CαΊ­p nhαΊ­t giΓ‘ cΖ‘ bαΊ£n cho tuαΊ§n tiαΊΏp theo
return pd.DataFrame(data)
def current_basics(symbol: str, curday: str) -> dict:
# Check if Finnhub is configured
if not FINNHUB_KEYS:
print(f"⚠️ Finnhub not configured, skipping financial basics for {symbol}")
return {}
# Thα»­ vα»›i tαΊ₯t cαΊ£ cΓ‘c Finnhub API keys
for api_key in FINNHUB_KEYS:
try:
client = finnhub.Client(api_key=api_key)
# ThΓͺm timeout cho Finnhub client
raw = client.company_basic_financials(symbol, "all")
if not raw["series"]:
continue
merged = defaultdict(dict)
for metric, vals in raw["series"]["quarterly"].items():
for v in vals:
merged[v["period"]][metric] = v["v"]
latest = max((p for p in merged if p <= curday), default=None)
if latest is None:
continue
d = dict(merged[latest])
d["period"] = latest
return d
except Exception as e:
print(f"Error getting basics for {symbol} with key {api_key[:8]}...: {e}")
time.sleep(2) # ThΓͺm delay trΖ°α»›c khi thα»­ key tiαΊΏp theo
continue
return {}
def attach_news(symbol: str, df: pd.DataFrame) -> pd.DataFrame:
news_col = []
for _, row in df.iterrows():
start = row["Start Date"].strftime("%Y-%m-%d")
end = row["End Date"].strftime("%Y-%m-%d")
time.sleep(2) # TΔƒng delay để trΓ‘nh rate limit
# Check if Finnhub is configured
if not FINNHUB_KEYS:
print(f"⚠️ Finnhub not configured, using mock news for {symbol}")
news_data = create_mock_news(symbol, start, end)
news_col.append(json.dumps(news_data))
continue
# Thα»­ vα»›i tαΊ₯t cαΊ£ cΓ‘c Finnhub API keys
news_data = []
for api_key in FINNHUB_KEYS:
try:
client = finnhub.Client(api_key=api_key)
weekly = client.company_news(symbol, _from=start, to=end)
weekly_fmt = [
{
"date" : datetime.fromtimestamp(n["datetime"]).strftime("%Y%m%d%H%M%S"),
"headline": n["headline"],
"summary" : n["summary"],
}
for n in weekly
]
weekly_fmt.sort(key=lambda x: x["date"])
news_data = weekly_fmt
break # Thành công, thoÑt khỏi loop
except Exception as e:
print(f"Error with Finnhub key {api_key[:8]}... for {symbol} from {start} to {end}: {e}")
time.sleep(3) # ThΓͺm delay trΖ°α»›c khi thα»­ key tiαΊΏp theo
continue
# NαΊΏu khΓ΄ng cΓ³ news data, tαΊ‘o mock news
if not news_data:
news_data = create_mock_news(symbol, start, end)
news_col.append(json.dumps(news_data))
df["News"] = news_col
return df
def create_mock_news(symbol: str, start: str, end: str) -> list:
"""TαΊ‘o mock news data khi API khΓ΄ng hoαΊ‘t Δ‘α»™ng"""
mock_news = [
{
"date": f"{start}120000",
"headline": f"{symbol} Shows Strong Performance in Recent Trading",
"summary": f"Company {symbol} has demonstrated resilience in the current market conditions with positive investor sentiment."
},
{
"date": f"{end}090000",
"headline": f"Analysts Maintain Positive Outlook for {symbol}",
"summary": f"Financial analysts continue to recommend {symbol} based on strong fundamentals and growth prospects."
}
]
return mock_news
# ---------- PROMPT CONSTRUCTION -------------------------------------------
def sample_news(news: list[str], k: int = 5) -> list[str]:
if len(news) <= k:
return news
return [news[i] for i in sorted(random.sample(range(len(news)), k))]
def make_prompt(symbol: str, df: pd.DataFrame, curday: str, use_basics=False) -> str:
# Thα»­ vα»›i tαΊ₯t cαΊ£ cΓ‘c Finnhub API keys để lαΊ₯y company profile
company_blurb = f"[Company Introduction]:\n{symbol} is a publicly traded company.\n"
if FINNHUB_KEYS:
for api_key in FINNHUB_KEYS:
try:
client = finnhub.Client(api_key=api_key)
prof = client.company_profile2(symbol=symbol)
company_blurb = (
f"[Company Introduction]:\n{prof['name']} operates in the "
f"{prof['finnhubIndustry']} sector ({prof['country']}). "
f"Founded {prof['ipo']}, market cap {prof['marketCapitalization']:.1f} "
f"{prof['currency']}; ticker {symbol} on {prof['exchange']}.\n"
)
break # Thành công, thoÑt khỏi loop
except Exception as e:
print(f"Error getting company profile for {symbol} with key {api_key[:8]}...: {e}")
time.sleep(2) # ThΓͺm delay trΖ°α»›c khi thα»­ key tiαΊΏp theo
continue
else:
print(f"⚠️ Finnhub not configured, using basic company info for {symbol}")
# Past weeks block
past_block = ""
for _, row in df.iterrows():
term = "increased" if row["End Price"] > row["Start Price"] else "decreased"
head = (f"From {row['Start Date']:%Y-%m-%d} to {row['End Date']:%Y-%m-%d}, "
f"{symbol}'s stock price {term} from "
f"{row['Start Price']:.2f} to {row['End Price']:.2f}.")
news_items = json.loads(row["News"])
summaries = [
f"[Headline] {n['headline']}\n[Summary] {n['summary']}\n"
for n in news_items
if not n["summary"].startswith("Looking for stock market analysis")
]
past_block += "\n" + head + "\n" + "".join(sample_news(summaries, 5))
# Optional basic financials
if use_basics:
basics = current_basics(symbol, curday)
if basics:
basics_txt = "\n".join(f"{k}: {v}" for k, v in basics.items() if k != "period")
basics_block = (f"\n[Basic Financials] (reported {basics['period']}):\n{basics_txt}\n")
else:
basics_block = "\n[Basic Financials]: not available\n"
else:
basics_block = "\n[Basic Financials]: not requested\n"
horizon = f"{curday} to {n_weeks_before(curday, -1)}"
final_user_msg = (
company_blurb
+ past_block
+ basics_block
+ f"\nBased on all information before {curday}, analyse positive "
"developments and potential concerns for {symbol}, then predict its "
f"price movement for next week ({horizon})."
)
return final_user_msg
# ---------- LLM CALL -------------------------------------------------------
def chat_completion(prompt: str,
model_name: str = FINGPT_MODEL_NAME,
temperature: float = 0.7,
stream: bool = False,
symbol: str = "STOCK") -> str:
# Check if FinGPT model is configured
if model is None or tokenizer is None:
print(f"⚠️ FinGPT model not configured, using mock response for {symbol}")
return create_mock_ai_response(symbol)
try:
# Kết hợp system prompt và user prompt
full_prompt = f"{SYSTEM_PROMPT}\n\n{prompt}"
# Tokenize input
inputs = tokenizer.encode(full_prompt, return_tensors="pt", max_length=MAX_LENGTH, truncation=True)
# Generate response
with torch.no_grad():
outputs = model.generate(
inputs,
max_length=inputs.shape[1] + 512, # Generate up to 512 new tokens
temperature=temperature,
do_sample=True,
top_p=0.9,
top_k=50,
pad_token_id=tokenizer.eos_token_id,
eos_token_id=tokenizer.eos_token_id,
no_repeat_ngram_size=2,
early_stopping=True
)
# Decode response
response = tokenizer.decode(outputs[0][inputs.shape[1]:], skip_special_tokens=True)
# Clean up response
response = response.strip()
if not response:
return create_mock_ai_response(symbol)
return response
except Exception as e:
print(f"Error with FinGPT model: {e}")
return create_mock_ai_response(symbol)
def create_mock_ai_response(symbol: str) -> str:
"""TαΊ‘o mock AI response khi FinGPT model khΓ΄ng hoαΊ‘t Δ‘α»™ng"""
return f"""
[Positive Developments]
β€’ Strong market position and brand recognition for {symbol}
β€’ Recent quarterly earnings showing growth potential
β€’ Positive analyst sentiment and institutional investor interest
β€’ Technological innovation and market expansion opportunities
[Potential Concerns]
β€’ Market volatility and economic uncertainty
β€’ Competitive pressures in the industry
β€’ Regulatory changes that may impact operations
β€’ Global economic factors affecting stock performance
[Prediction & Analysis]
Based on the current market conditions and company fundamentals, {symbol} is expected to show moderate growth over the next week. The stock may experience some volatility but should maintain an upward trend with a potential price increase of 2-5%. This prediction is based on current market sentiment and technical analysis patterns.
Note: This is a demonstration response using mock data. For real investment decisions, please consult with qualified financial professionals.
"""
# ---------- MAIN PREDICTION FUNCTION -----------------------------------------
def predict(symbol: str = "AAPL",
curday: str = today(),
n_weeks: int = 3,
use_basics: bool = False,
stream: bool = False) -> tuple[str, str]:
try:
steps = [n_weeks_before(curday, n) for n in range(n_weeks + 1)][::-1]
df = get_stock_data(symbol, steps)
df = attach_news(symbol, df)
prompt_info = make_prompt(symbol, df, curday, use_basics)
answer = chat_completion(prompt_info, stream=stream, symbol=symbol)
return prompt_info, answer
except Exception as e:
error_msg = f"Error in prediction: {str(e)}"
print(f"Prediction error: {e}") # Log the error for debugging
return error_msg, error_msg
# ---------- HUGGINGFACE SPACES INTERFACE -----------------------------------------
def hf_predict(symbol, n_weeks, use_basics):
# 1. get curday
curday = date.today().strftime("%Y-%m-%d")
# 2. call predict
prompt, answer = predict(
symbol=symbol.upper(),
curday=curday,
n_weeks=int(n_weeks),
use_basics=bool(use_basics),
stream=False
)
return prompt, answer
# ---------- GRADIO INTERFACE -----------------------------------------
def create_interface():
with gr.Blocks(
title="FinRobot Forecaster",
theme=gr.themes.Soft(),
css="""
.gradio-container {
max-width: 1200px !important;
margin: auto !important;
}
#model_prompt_textbox textarea {
overflow-y: auto !important;
max-height: none !important;
min-height: 400px !important;
resize: vertical !important;
white-space: pre-wrap !important;
word-wrap: break-word !important;
height: auto !important;
}
#model_prompt_textbox {
height: auto !important;
}
#analysis_results_textbox textarea {
overflow-y: auto !important;
max-height: none !important;
min-height: 400px !important;
resize: vertical !important;
white-space: pre-wrap !important;
word-wrap: break-word !important;
height: auto !important;
}
#analysis_results_textbox {
height: auto !important;
}
.textarea textarea {
overflow-y: auto !important;
max-height: 500px !important;
resize: vertical !important;
}
.textarea {
height: auto !important;
min-height: 300px !important;
}
.gradio-textbox {
height: auto !important;
max-height: none !important;
}
.gradio-textbox textarea {
height: auto !important;
max-height: none !important;
overflow-y: auto !important;
}
"""
) as demo:
gr.Markdown("""
# πŸ€– FinRobot Forecaster
**AI-powered stock market analysis and prediction using FinGPT financial language model**
This application analyzes stock market data, company news, and financial metrics using the specialized FinGPT model to provide comprehensive market insights and predictions.
⚠️ **Note**: Free API keys have daily rate limits. If you encounter errors, the app will use mock data for demonstration purposes.
""")
with gr.Row():
with gr.Column(scale=1):
symbol = gr.Textbox(
label="Stock Symbol",
value="AAPL",
placeholder="Enter stock symbol (e.g., AAPL, MSFT, GOOGL)",
info="Enter the ticker symbol of the stock you want to analyze"
)
n_weeks = gr.Slider(
1, 6,
value=3,
step=1,
label="Historical Weeks to Analyze",
info="Number of weeks of historical data to include in analysis"
)
use_basics = gr.Checkbox(
label="Include Basic Financials",
value=True,
info="Include basic financial metrics in the analysis"
)
btn = gr.Button(
"πŸš€ Run Analysis",
variant="primary"
)
with gr.Column(scale=2):
with gr.Tabs():
with gr.Tab("πŸ“Š Analysis Results"):
gr.Markdown("**AI Analysis & Prediction**")
output_answer = gr.Textbox(
label="",
lines=40,
show_copy_button=True,
interactive=False,
placeholder="AI analysis and predictions will appear here...",
container=True,
scale=1,
elem_id="analysis_results_textbox"
)
with gr.Tab("πŸ” Model Prompt"):
gr.Markdown("**Generated Prompt**")
output_prompt = gr.Textbox(
label="",
lines=40,
show_copy_button=True,
interactive=False,
placeholder="Generated prompt will appear here...",
container=True,
scale=1,
elem_id="model_prompt_textbox"
)
# Examples
gr.Examples(
examples=[
["AAPL", 3, False],
["MSFT", 4, True],
["GOOGL", 2, False],
["TSLA", 5, True],
["NVDA", 3, True]
],
inputs=[symbol, n_weeks, use_basics],
label="πŸ’‘ Try these examples"
)
# Event handlers
btn.click(
fn=hf_predict,
inputs=[symbol, n_weeks, use_basics],
outputs=[output_prompt, output_answer],
show_progress=True
)
# Footer
gr.Markdown("""
---
**Disclaimer**: This application is for educational and research purposes only.
The predictions and analysis provided should not be considered as financial advice.
Always consult with qualified financial professionals before making investment decisions.
""")
return demo
# ---------- MAIN EXECUTION -----------------------------------------
if __name__ == "__main__":
demo = create_interface()
demo.launch(
server_name="0.0.0.0",
server_port=7860,
share=False,
show_error=True,
debug=False,
quiet=True
)