File size: 3,146 Bytes
b10cd7f | 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 110 111 | import os
import pypinyin
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import FileResponse
import akshare as ak
import pandas as pd
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
_stock_cache = None
def load_stock_cache():
global _stock_cache
if _stock_cache is None:
try:
df = ak.stock_info_a_code_name()
_stock_cache = []
for _, row in df.iterrows():
code = str(row['code'])
name = str(row['name'])
py = "".join(pypinyin.lazy_pinyin(name, style=pypinyin.Style.FIRST_LETTER)).lower()
_stock_cache.append({
"code": code,
"name": name,
"pinyin": py
})
except Exception as e:
print("Error loading stock cache:", e)
def get_stock_name(symbol: str) -> str:
load_stock_cache()
if _stock_cache is not None:
symbol = str(symbol)
for item in _stock_cache:
if item["code"] == symbol:
return item["name"]
return ""
def get_prefix_symbol(symbol: str) -> str:
if len(symbol) == 8:
return symbol
if symbol.startswith('6'):
return 'sh' + symbol
return 'sz' + symbol
def format_stock_data(df: pd.DataFrame) -> list:
formatted_data = []
if df is None or df.empty:
return formatted_data
for _, row in df.iterrows():
date_val = row.get('date', '')
if hasattr(date_val, 'strftime'):
date_str = date_val.strftime('%Y-%m-%d')
else:
date_str = str(date_val).split(' ')[0]
formatted_data.append({
'date': date_str,
'open': float(row['open']),
'close': float(row['close']),
'low': float(row['low']),
'high': float(row['high'])
})
return formatted_data
@app.get("/")
def serve_index():
# 根路由直接返回 index.html
html_path = os.path.join(os.path.dirname(__file__), "index.html")
return FileResponse(html_path)
@app.get("/api/health")
def health_check():
return {"status": "ok"}
@app.get("/api/stock")
def get_stock(symbol: str):
full_symbol = get_prefix_symbol(symbol)
stock_name = get_stock_name(symbol)
try:
df = ak.stock_zh_a_daily(symbol=full_symbol)
data = format_stock_data(df)
return {"name": stock_name, "data": data}
except Exception as e:
raise HTTPException(status_code=500, detail=f"网络请求失败。原因: {str(e)}")
@app.get("/api/search")
def search_stock(q: str):
load_stock_cache()
if not _stock_cache:
return []
q_lower = q.lower()
results = []
for item in _stock_cache:
if q_lower in item["code"] or q_lower in item["name"].lower() or q_lower in item["pinyin"]:
results.append(item)
if len(results) >= 10:
break
return results
|