HKOpenDataDemo / app.py
bennylam's picture
Upload 2 files
90d6090 verified
Raw
History Blame Contribute Delete
3.02 kB
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.responses import FileResponse
import pandas as pd
import numpy as np
from sklearn.linear_model import LinearRegression
import httpx # 新增:用於呼叫政府 Open API
app = FastAPI()
# --- 現有的 AI 分析邏輯與數據 (保持不變) ---
data_sources = {
"housing": pd.DataFrame({
"year": [2018, 2019, 2020, 2021, 2022, 2023],
"hk_island_avg_sqft": [16000, 16500, 16200, 16800, 15500, 15000],
"kowloon_avg_sqft": [13000, 13500, 13300, 13700, 12800, 12500],
"nt_avg_sqft": [11000, 11400, 11200, 11600, 10800, 10500]
}),
"traffic": pd.DataFrame({
"month": [1, 2, 3, 4, 5, 6],
"mass_transit": [4500, 4200, 4600, 4700, 4800, 4850],
"bus_co": [3800, 3500, 3900, 3950, 4000, 4100],
"ferry": [120, 110, 130, 135, 140, 145]
}),
"population": pd.DataFrame({
"year": [2018, 2019, 2020, 2021, 2022, 2023],
"total_population": [7451000, 7500700, 7481800, 7401500, 7346900, 7498100]
})
}
def train_model(X, y):
model = LinearRegression()
model.fit(X, y)
return model
@app.get("/api/analyze/{category}")
def analyze_data(category: str):
if category == "housing":
df = data_sources["housing"]
X = df[['year']]
y = df['nt_avg_sqft']
model = train_model(X, y)
pred_2024 = model.predict([[2024]])[0]
return {"category": "房價分析 (地政總署數據結構)", "prediction": f"2024 年新界區預測平均呎價: ${pred_2024:.2f}", "status": "模型訓練完成 (Linear Regression)"}
elif category == "traffic":
df = data_sources["traffic"]
X = df[['month']]
y = df['mass_transit']
model = train_model(X, y)
pred_next_month = model.predict([[7]])[0]
return {"category": "交通流量 (運輸署數據結構)", "prediction": f"7月份集體運輸系統預測日均客流: {pred_next_month:.0f} 千人次", "status": "分析涵蓋集體運輸、巴士與渡輪清算數據"}
elif category == "population":
df = data_sources["population"]
X = df[['year']]
y = df['total_population']
model = train_model(X, y)
pred_2024 = model.predict([[2024]])[0]
return {"category": "人口趨勢 (統計處數據結構)", "prediction": f"2024 年預測總人口: {int(pred_2024)} 人", "status": "趨勢分析完成"}
return {"error": "未知的分類"}
# --- 新增:KMB 路線數據代理 API ---
@app.get("/api/kmb/routes")
async def get_kmb_routes():
url = "https://data.etabus.gov.hk/v1/transport/kmb/route/"
async with httpx.AsyncClient() as client:
try:
response = await client.get(url, timeout=10.0)
return response.json()
except Exception as e:
return {"error": f"無法獲取開放數據: {str(e)}", "data": []}
# 掛載靜態檔案
app.mount("/", StaticFiles(directory="static", html=True), name="static")