返回博客

如何在 Django 中构建货币转换器：完整的汇率 API 教程（2026）

V
Vlado Grigirov
May 12, 2026
Django Python Currency API Exchange Rates Tutorial Finexly

大多数涉及金钱的 Django 应用，迟早都会需要汇率——无论是用三种货币开发票、将 Stripe 入账折算回基础货币，还是在结账页面渲染本地化价格。正确的做法是接入一个Django 汇率 API，把 FX 当作一个已经解决的问题。错误的做法是把汇率硬编码在 settings.py 里，或者在每个请求里都调用上游。

这篇指南端到端讲解一个生产级集成。我们将使用 Finexly API、Django 的缓存框架（用于热点读取）、一个用于持久化的模型、一个用于定时刷新的 management command、Decimal 精度（不让浮点误差吞掉零钱）、通过 Django Ninja 暴露的 REST 端点、一个 htmx 驱动的 UI，以及配合 responses 的 pytest 测试（永远不碰网络），在 Django 5.x 中构建一个小型货币转换器。读完后你会得到一套可以直接放进任何 Django 项目的服务层架构——SaaS 计费、电商、会计，凡是钱跨境流动的场景都适用。

为什么专用货币 API 优于硬编码汇率

把汇率硬编码到 settings.py 里是第一个错误答案。第二个错误答案是在每个视图里都去调上游。一个用对了的专用货币 API 给你硬编码值给不了的四样东西：

  • 按需新鲜。汇率在工作日盘中持续变动。即便是 24 小时陈旧的汇率，对 USD/JPY 或 EUR/TRY 这类高波动对而言也可能移动 1–2%——足以抹平一个 SaaS 套餐的利润率。
  • 覆盖面广。Finexly 覆盖 170 多种货币，包括新兴市场和 CBDC 参考汇率。很多开源 Django 包附带的欧洲央行源大概只覆盖 32 种主要货币。只要有一个用阿根廷比索或土耳其里拉付款的用户，这个差距就重要。
  • 统一契约。latest、historical、convert 端点使用同一个 JSON 形态，而不是把三个上游源用胶带粘在一起。
  • 可预期的配额。一份可推理的限速文档，而不是"欧洲央行因为我们轮询太凶把我们 IP 封了"。

如果你还在权衡选项，我们的 2026 免费 vs 付费货币 API 对比 和 ExchangeRate-API vs CurrencyLayer vs Finexly 详细梳理了备选方案。

我们要构建什么

一个意见鲜明的小型 Django 模块，叫 fx，把五件事做好：

  1. 通过一个类型化的服务类从 Finexly 拉取任意基础货币的最新汇率。
  2. 把热点读取缓存在 Django 缓存里（生产用 Redis，开发用 locmem）。
  3. 把每次拉取持久化到 ExchangeRate 模型，即便上游抽风，结账也不会挂。
  4. 通过 management command + cron（或 Celery beat）定时刷新汇率。
  5. 暴露一个 JSON 转换端点和一个 htmx 驱动的 HTML 表单。

作为可复用 app 接入任意已有 Django 项目即可。

前置条件

你需要 Python 3.11+、Django 5.0+ 和一个 Finexly API key。在 Finexly 注册页 免费拿一个——免费套餐每月 1,000 次请求，对一个缓存得当的 MVP 来说绰绰有余。

第 1 步：搭建 Django 项目

python -m venv .venv
source .venv/bin/activate
pip install "django>=5.0,<6.0" httpx python-decouple django-ninja
django-admin startproject site .
python manage.py startapp fx

把新 app 加入 site/settings.py 的 INSTALLED_APPS：

INSTALLED_APPS = [
    "django.contrib.admin",
    "django.contrib.auth",
    "django.contrib.contenttypes",
    "django.contrib.sessions",
    "django.contrib.messages",
    "django.contrib.staticfiles",
    "fx",
]

我们用 httpx 而不是 requests，因为它给我们一个客户端就能搞定 timeout、通过 httpx.HTTPTransport(retries=...) 的重试，而且之后可以换成 async 客户端。我们用 python-decouple 管环境变量，用 django-ninja 做 REST 端点——两个都最小、不挡路。

第 2 步：配置环境变量

在项目根目录建一个 .env：

DEBUG=True
SECRET_KEY=change-me-in-production
FINEXLY_API_KEY=your_finexly_api_key_here
FINEXLY_BASE_URL=https://finexly.com/api/v1
FX_CACHE_TTL=900
FX_DEFAULT_BASE=USD

再接到 site/settings.py：

from decouple import config

SECRET_KEY = config("SECRET_KEY")
DEBUG = config("DEBUG", default=False, cast=bool)

FINEXLY_API_KEY = config("FINEXLY_API_KEY")
FINEXLY_BASE_URL = config("FINEXLY_BASE_URL", default="https://finexly.com/api/v1")
FX_CACHE_TTL = config("FX_CACHE_TTL", default=900, cast=int)
FX_DEFAULT_BASE = config("FX_DEFAULT_BASE", default="USD")

CACHES = {
    "default": {
        "BACKEND": "django.core.cache.backends.redis.RedisCache",
        "LOCATION": config("REDIS_URL", default="redis://127.0.0.1:6379/1"),
    }
}

开发环境没有 Redis 时，把 RedisCache 换成 LocMemCache——其他代码不用动。

15 分钟的 TTL（900 秒）对大多数面向消费者的应用是合理默认。高频交易要 sub-second；会计报表 6 小时的 TTL 都行。我们的 货币 API 缓存与错误处理最佳实践 一文讨论了如何为你的场景挑合适的数字。

第 3 步：ExchangeRate 模型

持久化重要，因为它在上游不可达时给你一个回退，并在财务来问"4 月 3 日那张发票用了什么汇率"时给你一条审计线。建 fx/models.py：

from decimal import Decimal
from django.db import models


class ExchangeRate(models.Model):
    base = models.CharField(max_length=3, db_index=True)
    quote = models.CharField(max_length=3, db_index=True)
    rate = models.DecimalField(max_digits=20, decimal_places=10)
    fetched_at = models.DateTimeField(auto_now_add=True, db_index=True)

    class Meta:
        indexes = [
            models.Index(fields=["base", "quote", "-fetched_at"]),
        ]
        constraints = [
            models.CheckConstraint(
                check=models.Q(rate__gt=Decimal("0")),
                name="rate_positive",
            ),
        ]

    def __str__(self) -> str:
        return f"{self.base}/{self.quote} = {self.rate} @ {self.fetched_at:%Y-%m-%d %H:%M}"

两点值得说一下设计取舍。第一，DecimalField，不是 FloatField——浮点误差会在数千次转换中累积，你不会想做那个给财务解释为什么每张发票账目对不上 $0.03 的工程师。十位小数对展示来说过头，但对 FX 存储来说是标准。第二，(base, quote, -fetched_at) 复合索引把"给我最新的 USD/EUR 汇率"变成一次 B-tree 查找，而不是一次排序。

执行 migration：

python manage.py makemigrations fx
python manage.py migrate

第 4 步：服务层

服务类是所有跟上游对话的逻辑所在。视图永远不直接调用 httpx。建 fx/services.py：

import logging
from decimal import Decimal
from typing import Mapping

import httpx
from django.conf import settings
from django.core.cache import cache

logger = logging.getLogger(__name__)


class ExchangeRateError(Exception):
    """Upstream fetch failed and no usable fallback exists."""


class ExchangeRateService:
    def __init__(
        self,
        api_key: str | None = None,
        base_url: str | None = None,
        cache_ttl: int | None = None,
        default_base: str | None = None,
    ) -> None:
        self.api_key = api_key or settings.FINEXLY_API_KEY
        self.base_url = base_url or settings.FINEXLY_BASE_URL
        self.cache_ttl = cache_ttl or settings.FX_CACHE_TTL
        self.default_base = default_base or settings.FX_DEFAULT_BASE

    def latest(self, base: str | None = None) -> Mapping[str, Decimal]:
        base = (base or self.default_base).upper()
        cache_key = f"fx:latest:{base}"

        cached = cache.get(cache_key)
        if cached is not None:
            return cached

        rates = self._fetch_latest(base)
        cache.set(cache_key, rates, self.cache_ttl)
        self._persist(base, rates)
        return rates

    def convert(self, amount: Decimal, frm: str, to: str) -> Decimal:
        frm, to = frm.upper(), to.upper()
        if frm == to:
            return amount
        rates = self.latest(frm)
        if to not in rates:
            raise ExchangeRateError(f"Currency {to} not found in {frm} rates")
        return (amount * rates[to]).quantize(Decimal("0.01"))

    def _fetch_latest(self, base: str) -> dict[str, Decimal]:
        url = f"{self.base_url}/latest"
        headers = {"Authorization": f"Bearer {self.api_key}"}
        transport = httpx.HTTPTransport(retries=2)
        try:
            with httpx.Client(transport=transport, timeout=8.0) as client:
                response = client.get(url, params={"base": base}, headers=headers)
                response.raise_for_status()
        except httpx.HTTPError as exc:
            logger.warning("Finexly fetch failed for %s: %s", base, exc)
            fallback = self._fallback_from_db(base)
            if fallback is not None:
                return fallback
            raise ExchangeRateError(str(exc)) from exc

        payload = response.json()
        rates = payload.get("rates")
        if not isinstance(rates, dict):
            raise ExchangeRateError("Malformed response: missing 'rates'")
        return {quote: Decimal(str(rate)) for quote, rate in rates.items()}

    def _fallback_from_db(self, base: str) -> dict[str, Decimal] | None:
        from .models import ExchangeRate

        recent = (
            ExchangeRate.objects.filter(base=base)
            .order_by("quote", "-fetched_at")
            .distinct("quote")
        )
        rates = {row.quote: row.rate for row in recent}
        return rates or None

    def _persist(self, base: str, rates: Mapping[str, Decimal]) -> None:
        from .models import ExchangeRate

        ExchangeRate.objects.bulk_create(
            [ExchangeRate(base=base, quote=q, rate=r) for q, r in rates.items()],
            batch_size=200,
        )

这里有很多内容，拆开说。latest() 是唯一的公开读路径：先查缓存，再回退到上游，最后写回库。convert() 处理同币种的边界情况（避免不必要地乘以 1 而引入舍入），并把结果量化到两位小数——这是展示层选择；存储时保留更多精度。_fallback_from_db 是生产安全网：如果 Finexly 返回 503，我们从自己的数据库里按 quote 提供最新的汇率。结账不会因为第三方抽风而挂掉。

httpx.HTTPTransport 上的 retries=2 自动捕获绝大多数瞬时 502/503。8 秒超时是保守的——小于 5 秒在跨大洲时会太紧，大于 10 秒则一个抖动的上游就会吃掉你的请求延迟预算。

第 5 步：转换视图

现在是视图。建 fx/views.py：

from decimal import Decimal, InvalidOperation

from django.http import HttpRequest, JsonResponse
from django.shortcuts import render
from django.views.decorators.http import require_http_methods

from .services import ExchangeRateError, ExchangeRateService


@require_http_methods(["GET", "POST"])
def convert_view(request: HttpRequest):
    context: dict = {"result": None}
    if request.method == "POST":
        service = ExchangeRateService()
        try:
            amount = Decimal(request.POST.get("amount", "0"))
            frm = request.POST.get("from", "USD")
            to = request.POST.get("to", "EUR")
            converted = service.convert(amount, frm, to)
            context.update({"result": converted, "from": frm, "to": to, "amount": amount})
        except (InvalidOperation, ExchangeRateError) as exc:
            context["error"] = str(exc)

    template = "fx/_result.html" if request.headers.get("HX-Request") else "fx/convert.html"
    return render(request, template, context)

检查 HX-Request 请求头是 htmx 模式：同一个视图，两个模板。普通 POST 返回整页；htmx 触发的 POST 只返回结果片段，htmx 把它换入 DOM。用户看到转换后的金额原地出现，而你一行 JavaScript 都没写。

第 6 步：htmx 模板

建 fx/templates/fx/convert.html：

{% load static %}
<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <title>Currency Converter</title>
  <script src="https://unpkg.com/htmx.org@1.9.10"></script>
  <style>
    body { font-family: system-ui, sans-serif; max-width: 480px; margin: 4rem auto; }
    input, select, button { padding: 0.5rem; font-size: 1rem; }
    .result { margin-top: 1rem; font-size: 1.4rem; }
  </style>
</head>
<body>
  <h1>Currency Converter</h1>
  <form hx-post="{% url 'fx:convert' %}" hx-target="#result" hx-swap="innerHTML">
    {% csrf_token %}
    <input type="number" name="amount" step="0.01" required placeholder="Amount">
    <input type="text" name="from" maxlength="3" required placeholder="USD" value="USD">
    <input type="text" name="to" maxlength="3" required placeholder="EUR" value="EUR">
    <button type="submit">Convert</button>
  </form>
  <div id="result">{% include "fx/_result.html" %}</div>
</body>
</html>

以及局部模板 fx/templates/fx/_result.html：

{% if error %}
  <p class="error">{{ error }}</p>
{% elif result %}
  <p class="result">{{ amount }} {{ from }} = <strong>{{ result }} {{ to }}</strong></p>
{% endif %}

在 fx/urls.py 里挂上：

from django.urls import path
from . import views

app_name = "fx"
urlpatterns = [
    path("convert/", views.convert_view, name="convert"),
]

以及 site/urls.py：

from django.urls import include, path

urlpatterns = [
    path("", include("fx.urls")),
]

第 7 步：用 Django Ninja 暴露 REST 端点

对于 JSON 消费者——你的移动端、后台 worker、前端 SPA——暴露一个类型化端点。建 fx/api.py：

from decimal import Decimal

from ninja import NinjaAPI, Schema
from ninja.errors import HttpError

from .services import ExchangeRateError, ExchangeRateService

api = NinjaAPI(title="FX API")


class ConvertOut(Schema):
    amount: Decimal
    from_: str
    to: str
    rate: Decimal
    result: Decimal


@api.get("/convert", response=ConvertOut)
def convert(request, amount: Decimal, from_: str, to: str):
    service = ExchangeRateService()
    try:
        rates = service.latest(from_)
        rate = rates[to.upper()]
        result = service.convert(amount, from_, to)
    except (ExchangeRateError, KeyError) as exc:
        raise HttpError(502, f"FX unavailable: {exc}") from exc
    return ConvertOut(amount=amount, from_=from_.upper(), to=to.upper(), rate=rate, result=result)

加到 site/urls.py：

from fx.api import api

urlpatterns = [
    path("", include("fx.urls")),
    path("api/", api.urls),
]

现在 GET /api/convert?amount=100&from_=USD&to=EUR 返回一个强类型的 JSON 负载。Django Ninja 在 /api/docs 免费生成 OpenAPI schema——当你的移动团队开始提问时很有用。

第 8 步：通过 management command 定时刷新

对于大多数应用，聪明的模式是周期性后台刷新，在用户到达前预热缓存。建 fx/management/commands/refresh_rates.py：

from django.core.management.base import BaseCommand

from fx.services import ExchangeRateError, ExchangeRateService


class Command(BaseCommand):
    help = "Refresh exchange rates from Finexly for the configured base currencies."

    def add_arguments(self, parser):
        parser.add_argument("--bases", nargs="+", default=["USD", "EUR", "GBP"])

    def handle(self, *args, **options):
        service = ExchangeRateService()
        for base in options["bases"]:
            try:
                rates = service.latest(base)
                self.stdout.write(self.style.SUCCESS(
                    f"{base}: refreshed {len(rates)} rates"
                ))
            except ExchangeRateError as exc:
                self.stderr.write(self.style.ERROR(f"{base}: {exc}"))

你还需要空的 fx/management/__init__.py 和 fx/management/commands/__init__.py，Django 才能发现这个命令。

用 cron（或如果你本来就跑 Celery，用 Celery beat）来调度：

*/15 * * * * cd /app && /app/.venv/bin/python manage.py refresh_rates --bases USD EUR GBP JPY

每 15 分钟刷新四个基础货币。每次刷新一次上游调用——4 次 × 96（一天的间隔数）= 每天 384 次调用，若降为每小时一次，则仍稳稳在 Finexly 每月 1,000 次免费额度内。若你需要小于 15 分钟的新鲜度，参考 价格方案 页面看更高档位配额。

第 9 步：永远不碰网络的测试

服务层存在的全部意义就是你能 mock 掉它。安装 responses 和 pytest-django：

pip install pytest pytest-django responses

建 pytest.ini：

[pytest]
DJANGO_SETTINGS_MODULE = site.settings
python_files = tests.py test_*.py *_tests.py

然后 fx/tests/test_services.py：

from decimal import Decimal

import pytest
import responses
from django.core.cache import cache

from fx.services import ExchangeRateError, ExchangeRateService


@pytest.fixture(autouse=True)
def _clear_cache():
    cache.clear()
    yield
    cache.clear()


@responses.activate
def test_latest_returns_rates_from_upstream(settings):
    responses.add(
        responses.GET,
        f"{settings.FINEXLY_BASE_URL}/latest",
        json={"base": "USD", "rates": {"EUR": 0.92, "GBP": 0.79}},
        status=200,
    )
    service = ExchangeRateService()
    rates = service.latest("USD")
    assert rates["EUR"] == Decimal("0.92")
    assert rates["GBP"] == Decimal("0.79")


@responses.activate
def test_convert_quantizes_to_cents(settings):
    responses.add(
        responses.GET,
        f"{settings.FINEXLY_BASE_URL}/latest",
        json={"base": "USD", "rates": {"EUR": Decimal("0.9234")}},
        status=200,
    )
    service = ExchangeRateService()
    result = service.convert(Decimal("100"), "USD", "EUR")
    assert result == Decimal("92.34")


@responses.activate
def test_upstream_failure_raises_when_no_fallback(settings):
    responses.add(
        responses.GET,
        f"{settings.FINEXLY_BASE_URL}/latest",
        status=503,
    )
    service = ExchangeRateService()
    with pytest.raises(ExchangeRateError):
        service.latest("USD")


def test_same_currency_returns_input():
    service = ExchangeRateService()
    assert service.convert(Decimal("100"), "USD", "USD") == Decimal("100")

跑起来：

pytest -q

四个测试，零网络调用，总耗时亚秒。这就是整套设计的意义。

生产注意事项

几件在 FX 上线后会咬 Django 应用一口的事：

  • 数据库持久化是你的回退。如果你跳过 _persist，上游一停服结账就挂。别跳。
  • 缓存击穿。如果 100 个请求同时命中冷缓存，你会发出 100 个上游调用。用 django-redis 的 get_or_set 加锁，或者如果你流量很低，就接受这个权衡。
  • 周末汇率漂移。FX 市场休市。多数 API（Finexly 在内）整个周末提供周五收盘价；你的汇率要等周一开盘才会动。如果你结算周末交易，把规则写进文档。
  • 审计线。ExchangeRate 表长得很快。加个最近 7 天的 partial 索引，或者在合规允许时把更早的行卷入月度汇总。

如果你想深入这些模式，我们的 货币 API 缓存与错误处理指南 详细讲了击穿保护、断路器和优雅降级。

常见问题

哪个汇率 API 最适合 Django？ 任何 REST API 都能用，但人体工学差别很大。Finexly 返回扁平的 {rates: {...}} 形态，干净地映射到 Python dict，支持 170 多种货币，自带每月 1,000 次免费调用——做 MVP 够了。Fixer 和 Open Exchange Rates 之类的备选有不同配额结构；我们的 对比文章 走过这些取舍。

Django 里货币汇率应该用 Decimal 还是 Float？ 存储一律 DecimalField，业务逻辑一律 Decimal。Float 引入的舍入误差会在数千次转换中累积；在金融代码里这些误差最终会变成用户可见 bug。性能代价可以忽略不计。

汇率应该多久刷新一次？ 看你的场景。面向消费者展示本地化价格的应用，15 分钟一次足够。会计报表每小时即可。交易或实时对冲场景需要 sub-second WebSocket 源——这个模式参考我们的 REST vs WebSocket 文章。

我能用 Django REST Framework 代替 Django Ninja 吗？ 可以。服务层与框架无关。把 fx/api.py 换成一个调用 ExchangeRateService().convert(...) 并把结果序列化的 DRF APIView 即可。模型、服务、management command、测试都不用动。

怎么处理 API 响应里没有的货币？ 两种模式。要么显式抛 ExchangeRateError 然后展示给用户（UX 更干净），要么走 USD 中转的 cross-rate（更宽容，但更难调试）。对多数应用来说，早抛错是对的。

免费试用 Finexly

准备把实时汇率集成进你的 Django 项目了吗？免费拿一个 Finexly API key——不需要信用卡。每月 1,000 次免费请求起步，覆盖 170 多种货币的实时汇率，干净映射到 Python 的 JSON 契约。需要时在我们的 价格方案 上随应用扩展。

Explore More

API Documentation

Full endpoint reference & examples
Currency Converter

Try live rates instantly
Pricing Plans

Free tier & production plans
VG

Vlado Grigirov

Senior Currency Markets Analyst & Financial Strategist

Vlado Grigirov is a senior currency markets analyst and financial strategist with over 14 years of experience in foreign exchange markets, cross-border finance, and currency risk management. He has wo...

View full profile →

分享这篇文章

推特 领英 电子邮件