目前想要把先前的hello web擴充成一個畫價格的互動式網頁,不過今天沒有成功搞定,仍在嘗試。
目前我把價格檔案的數據放在‵~/log/demo1/data.csv建立python 3.12的環境後 安裝套件:
pip install -r requirements.txt執行網頁:
python app_single.py`
目前打算使用plotly來畫互動價格圖表,價格數據的格式如下:
open_time,open,high,low,close,volume,close_time,quote_asset_volume,number_of_trades,taker_buy_base,taker_buy_quote
2025-01-01 00:00:00,93548.8,93599.9,93514.2,93599.9,71.187,2025-01-01 00:00:59.999,6658941.517,1660,35.553,3325936.3709
2025-01-01 00:01:00,93599.9,93637.7,93577.6,93637.7,39.526,2025-01-01 00:01:59.999,3699698.6602,1371,24.39,2282997.2048
2025-01-01 00:02:00,93637.7,93690.0,93614.2,93688.5,94.805,2025-01-01 00:02:59.999,8878464.6639,2376,60.198,5637719.356
2025-01-01 00:03:00,93688.5,93688.5,93626.4,93664.6,45.566,2025-01-01 00:03:59.999,4267675.7253,1379,16.074,1505371.2095
2025-01-01 00:04:00,93664.5,93668.3,93626.3,93648.4,90.52,2025-01-01 00:04:59.999,8477820.7404,1390,12.103,1133541.5099
2025-01-01 00:05:00,93648.3,93666.7,93637.5,93663.4,26.631,2025-01-01 00:05:59.999,2494061.1365,817,16.622,1556687.2046
2025-01-01 00:06:00,93663.4,93663.4,93576.9,93620.8,40.442,2025-01-01 00:06:59.999,3785548.1259,1388,20.196,1890172.6403
2025-01-01 00:07:00,93620.8,93620.8,93592.8,93592.8,22.117,2025-01-01 00:07:59.999,2070222.3945,688,6.306,590214.5918
flask==3.0.3
pandas==2.2.2
numpy==2.0.1
matplotlib==3.9.0
plotly==5.22.0
mplfinance==0.12.10b0
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import os
from pathlib import Path
from io import BytesIO
from typing import List
import numpy as np
import pandas as pd
from flask import Flask, Response, abort, send_file, render_template_string
# 可選:延遲載入 plotly;沒有則設旗標避免 500
try:
import plotly.graph_objs as go
HAS_PLOTLY = True
except Exception:
HAS_PLOTLY = False
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
LOG_ROOT = Path(os.environ.get("LOG_ROOT", str(Path.home() / "log"))).expanduser().resolve()
PRICE_CSV = "data.csv" # 固定檔名(照你要求)
app = Flask(__name__)
# ----------------- 工具 -----------------
def list_dirs(root: Path) -> List[Path]:
if not root.exists():
return []
return sorted([p for p in root.iterdir() if p.is_dir()], key=lambda p: p.name)
def parse_prices_raw(path: Path) -> pd.DataFrame:
# 原封不動讀檔(debug 用)
df = pd.read_csv(path, engine="python")
return df
def parse_prices_plotdf(path: Path) -> pd.DataFrame:
"""轉成作圖 DataFrame:index=close_time(open_time+1min),欄位 Open/High/Low/Close/Volume"""
df = pd.read_csv(path, engine="python")
need = {"open_time", "open", "high", "low", "close"}
missing = need - set(df.columns)
if missing:
raise ValueError(f"價格CSV缺欄位:{missing}")
# 時間處理
df["open_time"] = pd.to_datetime(df["open_time"], errors="coerce")
df = df.dropna(subset=["open_time"]).copy()
# 就算 CSV 有 close_time,也統一用 open_time + 1min
df["close_time"] = df["open_time"] + pd.Timedelta(minutes=1)
df = df.sort_values("close_time").reset_index(drop=True)
# 轉數字
for c in ["open", "high", "low", "close", "volume"]:
if c in df.columns:
df[c] = pd.to_numeric(df[c], errors="coerce")
if "volume" not in df.columns:
df["volume"] = 0.0
out = pd.DataFrame(
{
"Open": df["open"].to_numpy(float),
"High": df["high"].to_numpy(float),
"Low": df["low"].to_numpy(float),
"Close": df["close"].to_numpy(float),
"Volume":df["volume"].to_numpy(float),
},
index=pd.DatetimeIndex(df["close_time"]),
).replace([np.inf, -np.inf], np.nan).dropna()
if out.empty:
raise ValueError("價格資料為空(或全 NaN/Inf)。")
return out
# ----------------- 路由 -----------------
@app.route("/ping")
def ping():
return "ok", 200
@app.route("/")
def home():
items = list_dirs(LOG_ROOT)
if not items:
return f"""
<h3>找不到任何子資料夾:{LOG_ROOT}</h3>
<p>請確認本機 <code>{LOG_ROOT}</code> 底下至少有一個子資料夾,且其中包含:
<code>{PRICE_CSV}</code></p>
""", 200
def row(i, p):
return (f'<li>索引 {i}: <b>{p.name}</b> | '
f'<a href="/debug/{i}">debug</a> | '
f'<a href="/line/{i}.png">line</a> | '
f'<a href="/plotly/{i}">plotly</a></li>')
lis = [row(i, p) for i, p in enumerate(items)]
return f"""
<h2>DGT 單檔除錯版</h2>
<div>根目錄:<code>{LOG_ROOT}</code></div>
<ol>{"".join(lis)}</ol>
""", 200
@app.route("/debug/<int:idx>")
def debug_idx(idx: int):
items = list_dirs(LOG_ROOT)
if idx < 0 or idx >= len(items):
abort(404, description="索引超出範圍。")
folder = items[idx]
price_path = folder / PRICE_CSV
if not price_path.is_file():
return Response(f"[DEBUG] 找不到價格檔:{price_path}\n", mimetype="text/plain")
try:
df = parse_prices_raw(price_path)
lines = []
lines.append(f"path: {price_path}")
lines.append(f"shape: {df.shape}")
lines.append(f"columns: {list(df.columns)}")
lines.append("head(3):")
lines.append(df.head(3).to_string())
# 二階段轉換結果
try:
d2 = parse_prices_plotdf(price_path)
lines.append(f"plot_df.shape: {d2.shape}")
lines.append(f"plot_df.index.tz: {getattr(d2.index, 'tz', None)}")
lines.append(f"plot_df.range: {d2.index.min()} ~ {d2.index.max()}")
except Exception as e:
lines.append(f"parse_prices_plotdf ERROR: {e}")
return Response("\n".join(lines) + "\n", mimetype="text/plain")
except Exception as e:
return Response(f"[ERROR] debug 失敗:{e}\n", mimetype="text/plain"), 500
@app.route("/line/<int:idx>.png")
def plot_line(idx: int):
items = list_dirs(LOG_ROOT)
if idx < 0 or idx >= len(items):
abort(404, description="索引超出範圍。")
folder = items[idx]
price_path = folder / PRICE_CSV
if not price_path.is_file():
abort(404, description=f"找不到價格檔:{PRICE_CSV}")
try:
d2 = parse_prices_plotdf(price_path)
x = d2.index
y = d2["Close"].to_numpy(float)
fig = plt.figure(figsize=(12, 6))
ax = fig.add_subplot(111)
ax.plot(x, y, linewidth=1)
ax.set_title("Close (line)")
ax.set_xlabel("time")
ax.set_ylabel("price")
fig.tight_layout()
buf = BytesIO()
fig.savefig(buf, format="png", dpi=140)
plt.close(fig)
buf.seek(0)
return send_file(buf, mimetype="image/png")
except Exception as e:
return Response(f"[ERROR] line 失敗:{e}\n", mimetype="text/plain"), 500
@app.route("/plotly/<int:idx>")
def plotly_price(idx: int):
if not HAS_PLOTLY:
return Response(
"尚未安裝 plotly。請在虛擬環境執行:\n\n"
" pip install plotly==5.22.0\n\n"
"裝好後重跑本程式,再開此頁即可看到互動圖表。\n",
mimetype="text/plain"
)
items = list_dirs(LOG_ROOT)
if idx < 0 or idx >= len(items):
abort(404, description="索引超出範圍。")
folder = items[idx]
price_path = folder / PRICE_CSV
if not price_path.is_file():
abort(404, description=f"找不到價格檔:{PRICE_CSV}")
try:
d2 = parse_prices_plotdf(price_path)
# 建立圖表物件
fig = go.Figure()
fig.add_trace(go.Candlestick(
x=d2.index, open=d2["Open"], high=d2["High"],
low=d2["Low"], close=d2["Close"], name="K線"
))
fig.update_layout(
title=f"{idx}: {folder.name}",
xaxis=dict(title="時間", rangeslider=dict(visible=False)),
yaxis=dict(title="價格"),
hovermode="x unified",
margin=dict(l=60, r=20, t=60, b=40),
)
# ✅ 關鍵:轉為 JSON 字串(可被前端安全使用)
import plotly.io as pio
fig_json = pio.to_json(fig) # 這是一個 JSON 字串
# 直接用 inline HTML,避免外部模板檔
html = render_template_string("""
<!doctype html>
<meta charset="utf-8">
<title>DGT 互動 K 線</title>
<script src="https://cdn.plot.ly/plotly-latest.min.js"></script>
<div id="chart" style="width: 100%; max-width: 1200px; height: 72vh;"></div>
<script>
const fig = JSON.parse({{ fig_json|tojson|safe }});
Plotly.newPlot('chart', fig.data, fig.layout, {responsive: true, displaylogo: false});
</script>
""", fig_json=fig_json)
return html
except Exception as e:
return Response(f"[ERROR] plotly 失敗:{e}\n", mimetype="text/plain"), 500
# ----------------- 進入點 -----------------
if __name__ == "__main__":
# 本地直跑,不用 Docker
# 預設 http://127.0.0.1:5000
print(f"[INFO] LOG_ROOT = {LOG_ROOT}")
app.run(host="127.0.0.1", port=5000, debug=True)