|
|
from fastapi import FastAPI |
|
|
from fastapi.middleware.cors import CORSMiddleware |
|
|
from vnstock import Vnstock |
|
|
from datetime import datetime, timedelta |
|
|
import pandas as pd |
|
|
import numpy as np |
|
|
|
|
|
vn = Vnstock() |
|
|
app = FastAPI(title="mFund VNStock API", version="1.0") |
|
|
|
|
|
app.add_middleware( |
|
|
CORSMiddleware, |
|
|
allow_origins=["*"], |
|
|
allow_credentials=True, |
|
|
allow_methods=["*"], |
|
|
allow_headers=["*"], |
|
|
) |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def calc_RSI(series, period=14): |
|
|
delta = series.diff() |
|
|
gain = delta.clip(lower=0) |
|
|
loss = -delta.clip(upper=0) |
|
|
avg_gain = gain.rolling(period).mean() |
|
|
avg_loss = loss.rolling(period).mean() |
|
|
RS = avg_gain / avg_loss |
|
|
return 100 - (100 / (1 + RS)) |
|
|
|
|
|
def calc_MACD(series, fast=12, slow=26, signal=9): |
|
|
ema_fast = series.ewm(span=fast, adjust=False).mean() |
|
|
ema_slow = series.ewm(span=slow, adjust=False).mean() |
|
|
macd = ema_fast - ema_slow |
|
|
signal_line = macd.ewm(span=signal, adjust=False).mean() |
|
|
return macd, signal_line, macd - signal_line |
|
|
|
|
|
def calc_bollinger(series, window=20, num_std=2): |
|
|
sma = series.rolling(window).mean() |
|
|
std = series.rolling(window).std() |
|
|
return sma, sma + num_std * std, sma - num_std * std |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@app.get("/stock/history") |
|
|
def get_history(symbol: str, start: str, end: str): |
|
|
try: |
|
|
if end < start: |
|
|
return {"error": "end must be >= start"} |
|
|
|
|
|
stock = vn.stock(symbol=symbol) |
|
|
if stock is None: |
|
|
return {"error": "invalid symbol"} |
|
|
|
|
|
df = stock.quote.history(start=start, end=end, interval="1D") |
|
|
|
|
|
if df is None or df.empty: |
|
|
return {"error": "no data"} |
|
|
|
|
|
if "time" in df.columns: |
|
|
df = df.rename(columns={"time": "Date"}).set_index("Date") |
|
|
|
|
|
if "close" in df.columns: |
|
|
df = df.rename(columns={"close": "Close"}) |
|
|
|
|
|
df.index = df.index.astype(str) |
|
|
return {"symbol": symbol, "data": df.to_dict()} |
|
|
except Exception as e: |
|
|
return {"error": str(e)} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@app.get("/stock/ta") |
|
|
def get_ta(symbol: str, start: str, end: str): |
|
|
try: |
|
|
if end < start: |
|
|
return {"error": "end must be >= start"} |
|
|
|
|
|
stock = vn.stock(symbol=symbol) |
|
|
if stock is None: |
|
|
return {"error": "invalid symbol"} |
|
|
|
|
|
df = stock.quote.history(start=start, end=end, interval="1D") |
|
|
|
|
|
if df is None or df.empty: |
|
|
return {"error": "no data"} |
|
|
|
|
|
df["RSI"] = calc_RSI(df["close"]) |
|
|
df["MACD"], df["MACD_signal"], df["MACD_hist"] = calc_MACD(df["close"]) |
|
|
df["BB_MID"], df["BB_UPPER"], df["BB_LOWER"] = calc_bollinger(df["close"]) |
|
|
|
|
|
df = df.fillna(None) |
|
|
df.index = df.index.astype(str) |
|
|
return {"symbol": symbol, "indicators": df.to_dict()} |
|
|
except Exception as e: |
|
|
return {"error": str(e)} |
|
|
|
|
|
@app.get("/") |
|
|
def root(): |
|
|
return { |
|
|
"message": "mFund VNStock FastAPI is running", |
|
|
"endpoints": [ |
|
|
"/stock/history?symbol=FPT&start=2023-01-01&end=2023-12-31", |
|
|
"/stock/ta?symbol=HPG&start=2023-01-01&end=2023-12-31", |
|
|
], |
|
|
} |
|
|
|
