iT邦幫忙

2025 iThome 鐵人賽

DAY 11
0

目前想要把先前的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)

上一篇
Day 10 - 在ECS Cluster 建立 Service
下一篇
Day12 - 本地開發價格繪製網頁,Debug中2
系列文
從零開始:AWS 部署 Python 自動交易程式與交易監測 Dashboard 實戰筆記12
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言