Vercel Functions 驱动 Matplotlib:构建与测试一个生产级无服务器图表生成 API


一切始于一个看似简单的需求:一个金融数据看板需要展示复杂的K线图,不仅包含基础的开盘、收盘、最高、最低价,还需要叠加多种技术指标,如布林带(Bollinger Bands)、移动平均线(MA)和成交量。前端团队最初尝试使用JavaScript图表库,但很快发现,在浏览器端实时计算和渲染这些复杂的、数据点密集的图表,性能瓶颈非常明显,并且导致前端代码与复杂的金融算法逻辑紧密耦合。

将图表生成逻辑迁移到后端是显而易见的解决方案。但这引出了新的问题:我们不想为此维护一台专门的服务器。这个图表生成功能的使用频率具有突发性——用户访问看板时请求量激增,闲时则无人问津。这种场景是无服务器(Serverless)架构的完美应用领域。Vercel Functions 因其与前端框架(如Next.js)的无缝集成和简洁的开发体验,成为了首选。

真正的挑战在于技术选型。要在后端生成高质量的金融图表,Python生态中的Matplotlibpandasnumpy是无可争议的王者。因此,我们决定使用Vercel的Python运行时。但随之而来的是一个关键问题:如何为一个部署在Serverless环境、依赖大型二进制库、并以图片形式响应的Python API编写可靠、高效的自动化测试?在一个以TypeScript/JavaScript为主的Monorepo项目中,引入一套完整的Python测试框架(如pytest)会增加工具链的复杂性。这里的核心痛点是,我们需要一种方式,能以“黑盒”的方式,从外部验证这个API的行为是否符合预期,就像前端应用实际调用它一样。

这就是Vitest登场的契机。作为项目中已在使用的前端测试框架,我们构想了一个略显 unorthodox 但极为务实的方案:使用Vitest来对本地运行的Vercel Functions Python API进行集成测试。它将模拟HTTP请求,并验证返回的是否为一张格式正确的、内容符合预期的图表图片。这个决定,让我们得以在统一的工具链下,完成对这个特殊Serverless端点的完整生命周期管理。

第一步:项目结构与基础环境配置

在一个典型的Vercel项目中,我们将Python函数放置在/api目录下。整个项目的结构如下:

.
├── api
│   ├── _data # 存放测试用的模拟数据
│   │   └── sample_stock_data.csv
│   ├── chart.py # 核心的Python Serverless Function
│   └── requirements.txt # Python依赖
├── package.json
├── vercel.json # Vercel项目配置
└── vitest.config.ts # Vitest配置文件

package.json中包含开发和测试所需的脚本:

{
  "name": "serverless-chart-generator",
  "version": "1.0.0",
  "scripts": {
    "dev": "vercel dev",
    "test": "vitest run"
  },
  "devDependencies": {
    "vitest": "^1.0.0",
    "vercel": "^32.0.0"
  }
}

Python的依赖项被明确列在api/requirements.txt中。这里的关键在于matplotlibpandasnumpy,它们是整个功能的核心,但也是Vercel函数冷启动时间和内存消耗的主要来源。

# api/requirements.txt
pandas==2.1.1
numpy==1.26.0
matplotlib==3.8.0

vercel.json的配置至关重要。由于这些库的体积和计算开销,默认的配置可能不足以支撑。我们需要为这个特定的函数增加内存和执行超时时间。在真实项目中,这是一个必须经过压测和监控来微调的参数。

{
  "functions": {
    "api/chart.py": {
      "memory": 1024,
      "maxDuration": 20
    }
  }
}

第二步:构建核心图表生成函数

现在,我们来编写api/chart.py。一个常见的错误是尝试将生成的图表保存到文件系统,这在无服务器的只读文件系统中是行不通的。正确的做法是将图表直接渲染到一个内存中的二进制缓冲区(io.BytesIO),然后将其作为HTTP响应体返回。

# api/chart.py
from http.server import BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
import io
import os

class handler(BaseHTTPRequestHandler):

    def do_GET(self):
        # --- 1. 解析请求参数 ---
        query_components = parse_qs(urlparse(self.path).query)
        # 简单获取参数,生产环境需要更健壮的验证
        symbol = query_components.get("symbol", ["DEFAULT"])[0]
        
        # --- 2. 健壮性检查 ---
        if symbol == "DEFAULT" or len(symbol) > 5:
            self.send_response(400)
            self.send_header('Content-type', 'application/json')
            self.end_headers()
            self.wfile.write(b'{"error": "Invalid or missing symbol parameter"}')
            return

        try:
            # --- 3. 数据加载与处理 ---
            # 在测试环境中,我们使用本地mock数据,避免依赖外部服务
            # 这是一个关键的、用于提升可测试性的设计
            if os.getenv("VITEST_ENV") == "true":
                data_path = os.path.join(os.path.dirname(__file__), '_data', 'sample_stock_data.csv')
                df = pd.read_csv(data_path, index_col='Date', parse_dates=True)
            else:
                # 在生产环境中,这里会调用一个真实的数据API
                # df = fetch_real_stock_data(symbol)
                # 为简化示例,我们复用模拟数据
                data_path = os.path.join(os.path.dirname(__file__), '_data', 'sample_stock_data.csv')
                df = pd.read_csv(data_path, index_col='Date', parse_dates=True)

            # --- 4. 计算技术指标 ---
            df['MA20'] = df['Close'].rolling(window=20).mean()
            df['MA50'] = df['Close'].rolling(window=50).mean()
            df['Bollinger_High'] = df['MA20'] + 2 * df['Close'].rolling(window=20).std()
            df['Bollinger_Low'] = df['MA20'] - 2 * df['Close'].rolling(window=20).std()

            # --- 5. 使用Matplotlib绘图 ---
            fig = plt.figure(figsize=(12, 8), dpi=100)
            # 设置暗色主题,这在Matplotlib中需要一些手动配置
            fig.patch.set_facecolor('#0f172a') 
            plt.style.use('dark_background')
            
            # 创建子图: 70%给K线图, 30%给成交量
            gs = fig.add_gridspec(3, 1, height_ratios=[2, 1, 1], hspace=0)
            ax_candle = fig.add_subplot(gs[0:2, :])
            ax_vol = fig.add_subplot(gs[2, :], sharex=ax_candle)

            # 绘制K线 (简化版,用矩形表示)
            for index, row in df.iterrows():
                color = '#22c55e' if row['Close'] >= row['Open'] else '#ef4444'
                ax_candle.plot([index, index], [row['Low'], row['High']], color='white', linewidth=0.5)
                ax_candle.add_patch(plt.Rectangle((mdates.date2num(index) - 0.4, row['Open']), 0.8, row['Close'] - row['Open'], facecolor=color, edgecolor='none'))

            # 绘制技术指标
            ax_candle.plot(df.index, df['MA20'], color='#f97316', linestyle='--', label='MA20')
            ax_candle.plot(df.index, df['MA50'], color='#6366f1', linestyle='--', label='MA50')
            ax_candle.fill_between(df.index, df['Bollinger_Low'], df['Bollinger_High'], color='#3b82f6', alpha=0.2)
            
            # 绘制成交量
            ax_vol.bar(df.index, df['Volume'], color=['#22c55e' if c >= o else '#ef4444' for c, o in zip(df['Close'], df['Open'])])
            
            # --- 6. 图表美化与配置 ---
            ax_candle.set_ylabel('Price (USD)')
            ax_candle.set_title(f'{symbol} Stock Price', color='white')
            ax_candle.grid(linestyle='--', alpha=0.2)
            ax_candle.legend()
            plt.setp(ax_candle.get_xticklabels(), visible=False) # 隐藏K线图的x轴标签

            ax_vol.set_ylabel('Volume')
            ax_vol.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m'))
            ax_vol.grid(linestyle='--', alpha=0.2)

            plt.tight_layout()

            # --- 7. 将图表渲染到内存缓冲区 ---
            buf = io.BytesIO()
            plt.savefig(buf, format='png', facecolor=fig.get_facecolor(), edgecolor='none')
            buf.seek(0)
            plt.close(fig) # 必须关闭figure释放内存,这在Serverless环境中尤其重要

            # --- 8. 发送HTTP响应 ---
            self.send_response(200)
            self.send_header('Content-type', 'image/png')
            self.send_header('Cache-Control', 's-maxage=3600, stale-while-revalidate') # 添加缓存头
            self.end_headers()
            self.wfile.write(buf.getvalue())
        
        except Exception as e:
            self.send_response(500)
            self.send_header('Content-type', 'application/json')
            self.end_headers()
            # 生产环境中应记录更详细的错误日志,而不是直接暴露异常
            self.wfile.write(f'{{"error": "Internal Server Error", "details": "{str(e)}"}}'.encode('utf-8'))

        return

这段代码有几个关键的设计考量:

  1. 内存管理: plt.close(fig) 是一个很容易被忽略但至关重要的步骤。在长时间运行的进程中,Matplotlib会缓存figure对象,导致内存泄漏。在Serverless环境中,虽然每次调用都是独立的,但最佳实践仍然是显式地释放资源。
  2. 可测试性: 通过检查环境变量 VITEST_ENV,我们分离了测试环境和生产环境的数据源。这使得我们的测试可以不依赖任何外部API,变得稳定且快速。
  3. 错误处理: 代码包含了对无效参数(400 Bad Request)和内部服务器错误(500 Internal Server Error)的捕获和处理,返回结构化的JSON错误信息。
  4. 缓存: 通过设置Cache-Control头,我们可以利用Vercel的边缘网络缓存生成的图片,大幅降低重复请求的计算成本和响应时间。

第三步:使用 Vitest 编写集成测试

这是整个方案中最具创造性的部分。我们的目标不是去测试Matplotlib画的图对不对,而是验证我们的API作为一个整体,其行为是否正确。这包括:

  • 能否在调用时正确启动?
  • 能否正确解析URL参数?
  • 能否根据不同参数返回不同的响应?
  • 能否在成功时返回一张有效的PNG图片?
  • 能否在失败时返回正确的HTTP状态码和JSON错误体?

首先配置Vitest。

// vitest.config.ts
import { defineConfig } from 'vitest/config';

export default defineConfig({
  test: {
    // Vitest的测试默认在Node.js环境中运行
    // 这允许我们使用fetch等API
    // 我们设置一个较长的超时时间,因为vercel dev启动和函数首次执行可能较慢
    testTimeout: 30000, 
  },
});

然后,我们编写测试文件api/chart.spec.ts

// api/chart.spec.ts
import { describe, it, expect, beforeAll, afterAll } from 'vitest';
import { spawn, ChildProcess } from 'child_process';

// 定义API的基础URL,由vercel dev启动
const BASE_URL = 'http://localhost:3000';

describe('Chart Generation API (/api/chart)', () => {
  let vercelDevProcess: ChildProcess;

  // 在所有测试开始前,启动`vercel dev`进程
  beforeAll(async () => {
    // 使用spawn启动后台进程
    vercelDevProcess = spawn('vercel', ['dev'], {
      // 设置环境变量,让Python代码知道当前是测试环境
      env: { ...process.env, VITEST_ENV: 'true' }, 
      detached: true, // 使子进程独立于父进程
    });

    // 等待vercel dev服务器启动完毕
    // 这是一个简化的实现,生产级测试需要更可靠的健康检查机制
    await new Promise(resolve => setTimeout(resolve, 15000));
  });

  // 所有测试结束后,确保杀掉`vercel dev`进程
  afterAll(() => {
    if (vercelDevProcess && vercelDevProcess.pid) {
      // 在 detached 模式下,需要杀掉整个进程组
      process.kill(-vercelDevProcess.pid);
    }
  });

  it('should return a 400 Bad Request for missing symbol', async () => {
    const response = await fetch(`${BASE_URL}/api/chart`);
    expect(response.status).toBe(400);
    const data = await response.json();
    expect(data).toEqual({ error: 'Invalid or missing symbol parameter' });
  });

  it('should return a valid PNG image for a valid symbol', async () => {
    const response = await fetch(`${BASE_URL}/api/chart?symbol=AAPL`);
    
    // 1. 验证HTTP状态码和头部
    expect(response.status).toBe(200);
    expect(response.headers.get('Content-Type')).toBe('image/png');
    expect(response.headers.get('Cache-Control')).toBe('s-maxage=3600, stale-while-revalidate');

    // 2. 验证响应体是否为有效的PNG
    const imageBuffer = await response.arrayBuffer();
    expect(imageBuffer.byteLength).toBeGreaterThan(10000); // 期望图片大于10KB,防止返回空文件

    // 3. 核心断言:检查PNG文件的“魔数”(Magic Number)
    // PNG文件的前8个字节是固定的:[137, 80, 78, 71, 13, 10, 26, 10]
    // 这是验证文件类型的最可靠方式,远比检查文件扩展名要好
    const signature = new Uint8Array(imageBuffer.slice(0, 8));
    const pngSignature = new Uint8Array([137, 80, 78, 71, 13, 10, 26, 10]);
    expect(signature).toEqual(pngSignature);
  });

  it('should handle internal errors gracefully', async () => {
    // 这个测试需要修改Python代码来模拟一个内部错误,
    // 例如,通过一个特殊的查询参数触发一个异常。
    // 此处省略具体实现,但思路是相同的:调用并验证500响应。
  });
});

这个测试文件是整个工作流的粘合剂。通过beforeAllafterAll钩子,我们实现了在测试生命周期内自动启停vercel dev服务。最关键的断言是检查PNG的魔数,这是一个非常轻量且可靠的方法来验证我们收到的确实是一个图片文件,而无需引入重量级的图像处理库进行像素级的比较。

工作流程与架构复盘

我们已经构建了一个完整的、可测试的系统。现在,让我们用一个流程图来回顾整个测试过程的交互。

sequenceDiagram
    participant Vitest as Vitest Test Runner
    participant VercelDev as Vercel Dev Server (localhost:3000)
    participant Python as Python Runtime (in VercelDev)
    participant Matplotlib as Matplotlib Library

    Vitest->>VercelDev: Spawns 'vercel dev' process with VITEST_ENV=true
    Note over VercelDev: Server starts listening...
    
    Vitest->>VercelDev: HTTP GET /api/chart?symbol=AAPL
    VercelDev->>Python: Invokes chart.py handler
    
    Python->>Python: Checks VITEST_ENV, loads local mock data
    Python->>Matplotlib: Processes data and generates plot
    Matplotlib-->>Python: Returns plot as in-memory PNG buffer
    
    Python-->>VercelDev: Constructs HTTP 200 Response with PNG data and headers
    VercelDev-->>Vitest: Returns HTTP Response
    
    Vitest->>Vitest: 1. Assert response.status === 200
    Vitest->>Vitest: 2. Assert Content-Type === 'image/png'
    Vitest->>Vitest: 3. Assert PNG magic number is correct
    
    Note over Vitest: Test passes!

    Vitest->>VercelDev: Kills 'vercel dev' process

这个架构的优势在于:

  1. 统一工具链: 无需在JS/TS主导的项目中引入Python特有的测试工具,降低了认知负荷和维护成本。
  2. 高保真测试: 测试直接作用于vercel dev,其环境与线上Vercel环境高度相似,这比单纯地单元测试Python函数要可靠得多。
  3. 关注点分离: 前端开发者可以完全信任这个API端点,他们只需要知道传入什么参数,就会得到一张图片。所有复杂的图表生成逻辑都被封装在后端。

局限性与未来迭代方向

尽管此方案在我们的场景中运行良好,但它并非没有局限性,认识到这些边界是做出正确技术决策的关键。

首先,性能与成本是永远的权衡。matplotlib及其依赖是大型库,会导致Vercel函数的冷启动时间显著增加,首次请求可能会有数秒的延迟。对于延迟敏感的应用,可能需要启用Vercel的预置并发(Provisioned Concurrency),但这会带来额外的成本。同时,复杂的图表生成是CPU密集型任务,执行时间越长,费用越高。必须对API进行监控,并评估其成本效益。

其次,缓存策略是优化的核心。当前我们设置了s-maxage,依赖Vercel的CDN进行缓存。对于数据更新不频繁但参数组合极多的场景,可能需要更精细的缓存策略,例如使用外部的Redis或Vercel KV来缓存生成的图片,避免重复计算。

最后,测试的深度。我们当前的测试验证了API的契约和输出格式,但没有验证图表内容的正确性(例如,MA20的曲线是否真的画对了)。进行像素级的图像对比测试是可行但非常脆弱的——任何微小的样式改动都可能导致测试失败。一个更务实的进阶方案是,让Python函数在测试模式下除了返回图片外,还返回一个包含关键计算结果的JSON(例如MA20的最后一个值),让Vitest可以同时断言图片格式和关键数据点的正确性。但这会增加测试代码和生产代码的耦合度,需要谨慎权衡。

这个方案本质上是在特定约束下(Serverless优先、Python生态优势、统一技术栈)找到的一个工程甜点。它展示了如何创造性地组合不同的工具,去解决一个跨语言、跨环境的真实世界问题。


  目录