File size: 3,298 Bytes
b980261 d40eb94 b980261 d40eb94 b980261 d40eb94 09acea3 d40eb94 09acea3 d40eb94 09acea3 d40eb94 b980261 d40eb94 b980261 d40eb94 09acea3 b980261 09acea3 d40eb94 17ac23b d40eb94 09acea3 b980261 09acea3 d40eb94 365b31a 09acea3 d40eb94 b980261 d40eb94 09acea3 b980261 09acea3 d40eb94 17ac23b d40eb94 b980261 d40eb94 365b31a 09acea3 d40eb94 b980261 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 |
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=["*"],
)
# ============================
# Chỉ báo kỹ thuật
# ============================
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
# ============================
# 1) GET /stock/history
# ============================
@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)}
# ============================
# 2) GET /stock/ta
# ============================
@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",
],
}
|