98 lines
4.0 KiB
Python
98 lines
4.0 KiB
Python
import pandas as pd
|
|
import requests
|
|
import os
|
|
from datetime import datetime
|
|
from ta.trend import EMAIndicator
|
|
from ta.momentum import StochasticOscillator
|
|
|
|
class DataEngine:
|
|
def __init__(self, symbol, url):
|
|
self.symbol = symbol
|
|
self.url = url
|
|
self.cache_dir = "data_cache"
|
|
if not os.path.exists(self.cache_dir):
|
|
os.makedirs(self.cache_dir)
|
|
self.file_path = os.path.join(self.cache_dir, f"{self.symbol}.csv")
|
|
|
|
def fetch_data(self):
|
|
# 1. Try to load existing local data
|
|
local_df = pd.DataFrame()
|
|
if os.path.exists(self.file_path):
|
|
try:
|
|
local_df = pd.read_csv(self.file_path, parse_dates=['date'])
|
|
# If data is fresh (from today), return it immediately
|
|
if not local_df.empty and local_df['date'].max().date() >= datetime.today().date():
|
|
return local_df
|
|
except Exception as e:
|
|
print(f"Cache read error for {self.symbol}: {e}")
|
|
|
|
# 2. Fetch new data from URL
|
|
try:
|
|
headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'}
|
|
response = requests.get(self.url, headers=headers, timeout=10)
|
|
response.raise_for_status()
|
|
json_data = response.json()
|
|
|
|
# Parsing Logic (Handles JPM and generic JSON)
|
|
if isinstance(json_data, dict) and "historicalNAVList" in json_data:
|
|
new_df = pd.DataFrame(json_data["historicalNAVList"])
|
|
new_df = new_df.rename(columns={'navPrice': 'close'})
|
|
else:
|
|
new_df = pd.DataFrame(json_data)
|
|
|
|
new_df['date'] = pd.to_datetime(new_df['date'])
|
|
|
|
# 3. Merge and Save
|
|
if not local_df.empty:
|
|
# Combine, drop duplicates (keep newest), and sort
|
|
df = pd.concat([local_df, new_df]).drop_duplicates(subset=['date'], keep='last')
|
|
else:
|
|
df = new_df
|
|
|
|
df = df.sort_values('date').dropna(subset=['close'])
|
|
df.to_csv(self.file_path, index=False)
|
|
return df
|
|
|
|
except Exception as e:
|
|
print(f"Network error for {self.symbol}: {e}")
|
|
return local_df if not local_df.empty else None
|
|
|
|
def calculate_table_metrics(self, df):
|
|
if df is None or df.empty or len(df) < 2:
|
|
return None
|
|
|
|
last_close = float(df.iloc[-1]['close'])
|
|
prev_close = float(df.iloc[-2]['close'])
|
|
change_pct = ((last_close - prev_close) / prev_close) * 100
|
|
|
|
count = len(df)
|
|
df_52 = df.tail(min(count, 252))
|
|
|
|
# EMA Calculations (Returns % offset from Price)
|
|
def get_ema_offset(window):
|
|
if count >= window:
|
|
ema = EMAIndicator(close=df['close'], window=window).ema_indicator().iloc[-1]
|
|
return round(((last_close / ema) * 100) - 100, 1)
|
|
return "N/A"
|
|
|
|
# Stochastic Logic
|
|
k_val = d_val = "N/A"
|
|
if count >= 14:
|
|
# Note: Using rolling close as proxy for High/Low since many URLs only provide Close
|
|
rolling_high = df['close'].rolling(window=14).max()
|
|
rolling_low = df['close'].rolling(window=14).min()
|
|
stoch = StochasticOscillator(high=rolling_high, low=rolling_low, close=df['close'], window=14, smooth_window=3)
|
|
k_val = round(stoch.stoch().iloc[-1], 0)
|
|
d_val = round(stoch.stoch_signal().iloc[-1], 0)
|
|
|
|
return {
|
|
"last_close": round(last_close, 2),
|
|
"change_pct": round(change_pct, 2),
|
|
"low_52": round(float(df_52['close'].min()), 2),
|
|
"high_52": round(float(df_52['close'].max()), 2),
|
|
"last_ema20": get_ema_offset(20),
|
|
"last_ema50": get_ema_offset(50),
|
|
"last_ema100": get_ema_offset(100),
|
|
"last_ema200": get_ema_offset(200),
|
|
"kd_values": f"{k_val}/{d_val}" if k_val != "N/A" else "N/A"
|
|
} |