|
from fastapi.responses import HTMLResponse |
|
from fastapi.templating import Jinja2Templates |
|
from fastapi import FastAPI, Request, HTTPException |
|
from fastapi.middleware.cors import CORSMiddleware |
|
from utils import StockDataFetcher |
|
|
|
app = FastAPI(swagger_ui_parameters={"syntaxHighlight.theme": "obsidian"}) |
|
|
|
fetcher = StockDataFetcher() |
|
|
|
origins = ["*"] |
|
|
|
app.add_middleware( |
|
CORSMiddleware, |
|
allow_origins=origins, |
|
allow_credentials=True, |
|
allow_methods=["*"], |
|
allow_headers=["*"], |
|
) |
|
|
|
templates = Jinja2Templates(directory="templates") |
|
|
|
@app.get("/", response_class=HTMLResponse) |
|
async def read_root(request: Request): |
|
return templates.TemplateResponse("hello.html", {"request": request}) |
|
|
|
|
|
@app.get('/ltp') |
|
async def get_data(ticker: str): |
|
try: |
|
response = fetcher.fetch_latest_price(ticker) |
|
return {'ltp' :response} |
|
except: |
|
return {"Timeout" : "Error"} |
|
|
|
@app.get('/historical') |
|
async def get_stocks_data(ticker: str, intervals: int, days: int): |
|
try: |
|
response = fetcher.fetch_stock_data(ticker, 15, 10).to_dict(orient="records") |
|
return {"data" : response} |
|
except: |
|
return {"Timeout" : "Error"} |
|
|
|
@app.get('/chain') |
|
async def get_chain(ticker: str): |
|
try: |
|
response, exp = fetcher.fetch_option_chain(ticker) |
|
response = response.to_dict(orient="records") |
|
print(response, exp) |
|
return {"data" : response} |
|
except: |
|
return {"Timeout" : "Error"} |
|
|
|
@app.get('/search') |
|
async def stock_search(ticker: str): |
|
try: |
|
response = fetcher.search_entity(ticker) |
|
return {"data" : response} |
|
except: |
|
return {"Timeout" : "Error"} |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@app.get('/signal') |
|
async def get_stock_signal(index: str): |
|
try: |
|
response = fetcher.realtime_signal(index).tail(1).to_dict(orient="records")[0] |
|
return {"data" : response} |
|
except: |
|
return {"Timeout" : "Error"} |
