SPEC-TASAS-CAMBIO-AUTOMATICAS
Información del Documento
| Campo |
Valor |
| Código Gap |
GAP-MGN-003-001 |
| Módulo |
MGN-003 (Configuración Multi-moneda) |
| Título |
Actualización Automática de Tasas de Cambio |
| Prioridad |
P1 |
| Complejidad |
Media |
| Referencia Odoo |
base, currency_rate_live |
1. Resumen Ejecutivo
1.1 Descripción del Gap
El sistema requiere actualización automática de tasas de cambio desde proveedores externos (BCE, Banxico, Fed, etc.) para mantener conversiones precisas sin intervención manual diaria.
1.2 Justificación de Negocio
- Precisión financiera: Tasas actualizadas para valoraciones correctas
- Eficiencia operativa: Eliminación de actualización manual diaria
- Cumplimiento regulatorio: Uso de tasas oficiales de bancos centrales
- Multi-mercado: Soporte para operaciones en múltiples países
1.3 Alcance
- Integración con proveedores de tasas (BCE, Banxico, BCB, etc.)
- Programación de actualizaciones automáticas (cron)
- Histórico de tasas con auditoría
- Validación y alertas de anomalías
- Cache inteligente para rendimiento
2. Análisis de Referencia (Odoo)
2.1 Modelo res.currency
# Odoo: odoo/addons/base/models/res_currency.py
class Currency(models.Model):
_name = 'res.currency'
name = fields.Char(required=True) # ISO 4217 code
symbol = fields.Char()
rate = fields.Float(compute='_compute_current_rate', digits=(12, 6))
rate_ids = fields.One2many('res.currency.rate', 'currency_id')
rounding = fields.Float(default=0.01)
decimal_places = fields.Integer(compute='_compute_decimal_places')
active = fields.Boolean(default=True)
def _compute_current_rate(self):
"""Obtiene tasa vigente para fecha actual o específica"""
date = self._context.get('date') or fields.Date.today()
company = self.env.company
for currency in self:
currency.rate = self.env['res.currency.rate'].search([
('currency_id', '=', currency.id),
('name', '<=', date),
('company_id', '=', company.id),
], order='name desc', limit=1).rate or 1.0
2.2 Modelo res.currency.rate
# Odoo: odoo/addons/base/models/res_currency.py
class CurrencyRate(models.Model):
_name = 'res.currency.rate'
_order = 'name desc'
name = fields.Date(default=fields.Date.today, required=True) # Effective date
rate = fields.Float(digits=(12, 6), required=True, default=1.0)
currency_id = fields.Many2one('res.currency', required=True)
company_id = fields.Many2one('res.company')
_sql_constraints = [
('unique_name_per_day', 'unique(name, currency_id, company_id)',
'Only one rate per day per currency per company!')
]
2.3 Proveedores de Tasas (currency_rate_live)
# Odoo: odoo/addons/currency_rate_live/models/res_config_settings.py
CURRENCY_PROVIDER_SELECTION = [
('ecb', 'European Central Bank'),
('fta', 'Federal Tax Administration (Switzerland)'),
('banxico', 'Bank of Mexico'),
('boc', 'Bank of Canada'),
('xe_com', 'xe.com'),
('bnr', 'National Bank of Romania'),
('mindicador', 'Chilean mindicador.cl'),
('bcrp', 'Bank of Peru'),
('cbuae', 'UAE Central Bank'),
('nbp', 'National Bank of Poland'),
('cbr', 'Central Bank of Russia'),
('tcmb', 'Central Bank of Turkey'),
('bcb', 'Central Bank of Brazil'),
]
class ResCompany(models.Model):
_inherit = 'res.company'
currency_provider = fields.Selection(CURRENCY_PROVIDER_SELECTION)
currency_interval_unit = fields.Selection([
('manually', 'Manually'),
('daily', 'Daily'),
('weekly', 'Weekly'),
('monthly', 'Monthly'),
], default='manually')
currency_next_execution_date = fields.Date()
2.4 Métodos de Actualización
# Ejemplo: Proveedor ECB
def _parse_ecb_data(self, available_currencies):
"""Parse XML from European Central Bank"""
url = 'https://www.ecb.europa.eu/stats/eurofxref/eurofxref-daily.xml'
response = requests.get(url, timeout=30)
root = ET.fromstring(response.content)
data = {}
for cube in root.iter('{http://www.ecb.int/vocabulary/2002-08-01/eurofxref}Cube'):
currency = cube.attrib.get('currency')
rate = cube.attrib.get('rate')
if currency and rate:
if currency in available_currencies:
data[currency] = float(rate)
data['EUR'] = 1.0
return data
# Ejemplo: Proveedor Banxico
def _parse_banxico_data(self, available_currencies):
"""Parse data from Bank of Mexico SOAP service"""
url = 'https://www.banxico.org.mx/SieAPIRest/service/v1/series/SF43718/datos/oportuno'
headers = {'Bmx-Token': self.env.company.banxico_token}
response = requests.get(url, headers=headers, timeout=30)
data = response.json()
rate = float(data['bmx']['series'][0]['datos'][0]['dato'])
return {'USD': 1.0, 'MXN': rate}
2.5 Scheduled Action (ir.cron)
<!-- Odoo: currency_rate_live/data/currency_rate_cron.xml -->
<record id="ir_cron_currency_rate_update" model="ir.cron">
<field name="name">Currency Rate Update</field>
<field name="model_id" ref="base.model_res_company"/>
<field name="state">code</field>
<field name="code">model._cron_currency_rate_update()</field>
<field name="interval_number">1</field>
<field name="interval_type">days</field>
<field name="numbercall">-1</field>
</record>
3. Especificación Técnica
3.1 Modelo de Datos
-- =====================================================
-- EXTENSIÓN DE MONEDAS
-- =====================================================
-- Información adicional de moneda
CREATE TABLE currency_configs (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
currency_id UUID NOT NULL REFERENCES currencies(id) ON DELETE CASCADE,
tenant_id UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
-- Configuración de decimales
decimal_places INTEGER NOT NULL DEFAULT 2,
rounding DECIMAL(10, 6) NOT NULL DEFAULT 0.01,
-- Símbolo y formato
symbol VARCHAR(10),
symbol_position VARCHAR(10) DEFAULT 'after' CHECK (symbol_position IN ('before', 'after')),
thousands_separator VARCHAR(3) DEFAULT ',',
decimal_separator VARCHAR(3) DEFAULT '.',
-- Estado
is_active BOOLEAN NOT NULL DEFAULT TRUE,
-- Auditoría
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
created_by UUID REFERENCES users(id),
updated_by UUID REFERENCES users(id),
CONSTRAINT uq_currency_config_tenant UNIQUE (currency_id, tenant_id)
);
-- =====================================================
-- TASAS DE CAMBIO
-- =====================================================
-- Tasas históricas de cambio
CREATE TABLE currency_rates (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
currency_id UUID NOT NULL REFERENCES currencies(id) ON DELETE CASCADE,
tenant_id UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
-- Fecha efectiva y tasa
effective_date DATE NOT NULL,
rate DECIMAL(20, 10) NOT NULL CHECK (rate > 0),
inverse_rate DECIMAL(20, 10) GENERATED ALWAYS AS (1.0 / rate) STORED,
-- Origen de la tasa
source rate_source_type NOT NULL DEFAULT 'manual',
provider_id UUID REFERENCES currency_rate_providers(id),
provider_reference VARCHAR(100),
-- Auditoría
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
created_by UUID REFERENCES users(id),
-- Restricción de unicidad: una tasa por día por moneda por tenant
CONSTRAINT uq_currency_rate_day UNIQUE (currency_id, tenant_id, effective_date)
);
-- Tipo de origen de tasa
CREATE TYPE rate_source_type AS ENUM (
'manual', -- Ingresada manualmente
'automatic', -- Descargada automáticamente
'imported', -- Importada desde archivo
'calculated' -- Calculada (cross-rate)
);
-- Índices para búsqueda eficiente
CREATE INDEX idx_currency_rates_lookup
ON currency_rates(currency_id, tenant_id, effective_date DESC);
CREATE INDEX idx_currency_rates_date
ON currency_rates(effective_date DESC);
-- =====================================================
-- PROVEEDORES DE TASAS
-- =====================================================
CREATE TABLE currency_rate_providers (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
-- Identificación
code VARCHAR(50) NOT NULL UNIQUE,
name VARCHAR(200) NOT NULL,
description TEXT,
-- Tipo y configuración
provider_type provider_type_enum NOT NULL,
base_currency_id UUID NOT NULL REFERENCES currencies(id),
-- Endpoint de API
api_url VARCHAR(500) NOT NULL,
api_method VARCHAR(10) DEFAULT 'GET',
api_headers JSONB DEFAULT '{}',
response_format VARCHAR(20) DEFAULT 'json' CHECK (response_format IN ('json', 'xml', 'csv')),
-- Mapeo de respuesta (JSONPath o XPath)
rate_path VARCHAR(200), -- Path al valor de tasa
currency_path VARCHAR(200), -- Path al código de moneda
date_path VARCHAR(200), -- Path a la fecha
-- Monedas soportadas
supported_currencies TEXT[], -- Array de códigos ISO
-- Límites y throttling
rate_limit_per_minute INTEGER DEFAULT 60,
rate_limit_per_day INTEGER DEFAULT 1000,
-- Estado
is_active BOOLEAN NOT NULL DEFAULT TRUE,
last_success_at TIMESTAMPTZ,
last_error_at TIMESTAMPTZ,
last_error_message TEXT,
consecutive_failures INTEGER DEFAULT 0,
-- Auditoría
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
-- Tipos de proveedor
CREATE TYPE provider_type_enum AS ENUM (
'ecb', -- European Central Bank
'banxico', -- Banco de México
'bcb', -- Banco Central do Brasil
'bcrp', -- Banco Central de Reserva del Perú
'boc', -- Bank of Canada
'fed', -- Federal Reserve
'nbp', -- National Bank of Poland
'fixer', -- Fixer.io (comercial)
'openexchange', -- Open Exchange Rates
'xe', -- XE.com
'custom' -- Proveedor personalizado
);
-- Proveedores predefinidos
INSERT INTO currency_rate_providers (code, name, provider_type, base_currency_id, api_url, response_format, rate_path, currency_path) VALUES
('ecb', 'European Central Bank', 'ecb',
(SELECT id FROM currencies WHERE iso_code = 'EUR'),
'https://www.ecb.europa.eu/stats/eurofxref/eurofxref-daily.xml',
'xml', '//Cube[@currency]/@rate', '//Cube[@currency]/@currency'),
('banxico', 'Banco de México', 'banxico',
(SELECT id FROM currencies WHERE iso_code = 'MXN'),
'https://www.banxico.org.mx/SieAPIRest/service/v1/series/{series}/datos/oportuno',
'json', '$.bmx.series[0].datos[0].dato', NULL),
('bcb', 'Banco Central do Brasil', 'bcb',
(SELECT id FROM currencies WHERE iso_code = 'BRL'),
'https://olinda.bcb.gov.br/olinda/servico/PTAX/versao/v1/odata/CotacaoDolarDia(dataCotacao=@date)?@date=%27{date}%27&$format=json',
'json', '$.value[0].cotacaoCompra', NULL);
-- =====================================================
-- CONFIGURACIÓN POR TENANT
-- =====================================================
CREATE TABLE tenant_currency_settings (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
tenant_id UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
-- Moneda base del tenant
base_currency_id UUID NOT NULL REFERENCES currencies(id),
-- Proveedor de tasas preferido
preferred_provider_id UUID REFERENCES currency_rate_providers(id),
fallback_provider_id UUID REFERENCES currency_rate_providers(id),
-- Programación de actualización
update_interval update_interval_type NOT NULL DEFAULT 'daily',
update_time TIME DEFAULT '06:00:00', -- Hora local de actualización
update_timezone VARCHAR(50) DEFAULT 'UTC',
next_scheduled_update TIMESTAMPTZ,
-- Credenciales de API (encriptadas)
api_credentials JSONB DEFAULT '{}',
-- Configuración de validación
max_rate_change_percent DECIMAL(5, 2) DEFAULT 10.00, -- Alerta si cambia >10%
require_approval_above DECIMAL(5, 2) DEFAULT 20.00, -- Requiere aprobación si >20%
-- Historial
keep_history_days INTEGER DEFAULT 365, -- Días a mantener historial
-- Auditoría
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
CONSTRAINT uq_tenant_currency_settings UNIQUE (tenant_id)
);
-- Intervalo de actualización
CREATE TYPE update_interval_type AS ENUM (
'manually', -- Solo manual
'hourly', -- Cada hora
'daily', -- Diario
'weekly', -- Semanal (lunes)
'monthly' -- Mensual (día 1)
);
-- =====================================================
-- LOG DE ACTUALIZACIONES
-- =====================================================
CREATE TABLE currency_rate_update_logs (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
tenant_id UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
provider_id UUID NOT NULL REFERENCES currency_rate_providers(id),
-- Resultado
status update_status_type NOT NULL,
started_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
completed_at TIMESTAMPTZ,
duration_ms INTEGER,
-- Detalles
currencies_requested INTEGER DEFAULT 0,
currencies_updated INTEGER DEFAULT 0,
currencies_failed INTEGER DEFAULT 0,
-- Errores
error_code VARCHAR(50),
error_message TEXT,
error_details JSONB,
-- Request/Response para debugging
request_url TEXT,
request_headers JSONB,
response_status INTEGER,
response_body_sample TEXT, -- Primeros 1000 chars
-- Índices para limpieza
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE TYPE update_status_type AS ENUM (
'pending',
'in_progress',
'success',
'partial_success',
'failed',
'timeout',
'rate_limited'
);
-- Índice para limpieza de logs antiguos
CREATE INDEX idx_rate_update_logs_cleanup
ON currency_rate_update_logs(tenant_id, created_at);
-- =====================================================
-- ALERTAS DE TASAS
-- =====================================================
CREATE TABLE currency_rate_alerts (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
tenant_id UUID NOT NULL REFERENCES tenants(id) ON DELETE CASCADE,
currency_id UUID NOT NULL REFERENCES currencies(id),
-- Tipo de alerta
alert_type rate_alert_type NOT NULL,
-- Valores detectados
previous_rate DECIMAL(20, 10),
new_rate DECIMAL(20, 10),
change_percent DECIMAL(10, 4),
-- Estado
status alert_status_type NOT NULL DEFAULT 'pending',
requires_approval BOOLEAN DEFAULT FALSE,
-- Resolución
resolved_at TIMESTAMPTZ,
resolved_by UUID REFERENCES users(id),
resolution_action VARCHAR(50),
resolution_notes TEXT,
-- Auditoría
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
CREATE TYPE rate_alert_type AS ENUM (
'large_change', -- Cambio mayor al umbral
'no_data', -- Sin datos del proveedor
'stale_rate', -- Tasa no actualizada en X días
'provider_error', -- Error del proveedor
'validation_failed', -- Validación de negocio fallida
'manual_override' -- Tasa sobrescrita manualmente
);
CREATE TYPE alert_status_type AS ENUM (
'pending',
'acknowledged',
'approved',
'rejected',
'auto_resolved'
);
3.2 Servicios de Dominio
# app/modules/multi_currency/domain/services/currency_rate_service.py
from datetime import date, datetime, time, timedelta
from decimal import Decimal
from typing import Optional, Dict, List
from enum import Enum
import httpx
import asyncio
from xml.etree import ElementTree as ET
from app.core.exceptions import DomainException
from app.core.logging import get_logger
logger = get_logger(__name__)
class RateSourceType(str, Enum):
MANUAL = "manual"
AUTOMATIC = "automatic"
IMPORTED = "imported"
CALCULATED = "calculated"
class CurrencyRateService:
"""Servicio de dominio para gestión de tasas de cambio."""
def __init__(
self,
rate_repository: "CurrencyRateRepository",
provider_repository: "CurrencyRateProviderRepository",
settings_repository: "TenantCurrencySettingsRepository",
alert_service: "CurrencyRateAlertService",
cache_service: "CacheService"
):
self._rate_repo = rate_repository
self._provider_repo = provider_repository
self._settings_repo = settings_repository
self._alert_service = alert_service
self._cache = cache_service
async def get_rate(
self,
tenant_id: str,
currency_id: str,
target_date: Optional[date] = None,
use_cache: bool = True
) -> Decimal:
"""
Obtiene la tasa de cambio vigente para una moneda en una fecha.
Args:
tenant_id: ID del tenant
currency_id: ID de la moneda
target_date: Fecha para la tasa (default: hoy)
use_cache: Si usar cache (default: True)
Returns:
Tasa de cambio (rate relativo a moneda base del tenant)
"""
target_date = target_date or date.today()
# Verificar cache
if use_cache:
cache_key = f"rate:{tenant_id}:{currency_id}:{target_date}"
cached = await self._cache.get(cache_key)
if cached:
return Decimal(cached)
# Buscar tasa vigente (la más reciente <= target_date)
rate = await self._rate_repo.find_effective_rate(
tenant_id=tenant_id,
currency_id=currency_id,
effective_date=target_date
)
if not rate:
# Verificar si es la moneda base
settings = await self._settings_repo.get_by_tenant(tenant_id)
if settings and settings.base_currency_id == currency_id:
return Decimal("1.0")
raise DomainException(
code="RATE_NOT_FOUND",
message=f"No exchange rate found for currency {currency_id} on {target_date}"
)
# Guardar en cache (expira a medianoche)
if use_cache:
ttl = self._calculate_ttl_to_midnight()
await self._cache.set(cache_key, str(rate.rate), ttl=ttl)
return rate.rate
async def get_rate_between(
self,
tenant_id: str,
from_currency_id: str,
to_currency_id: str,
target_date: Optional[date] = None
) -> Decimal:
"""
Obtiene la tasa de conversión entre dos monedas.
Usa cross-rate a través de la moneda base si es necesario.
"""
target_date = target_date or date.today()
# Obtener tasas de ambas monedas respecto a moneda base
from_rate = await self.get_rate(tenant_id, from_currency_id, target_date)
to_rate = await self.get_rate(tenant_id, to_currency_id, target_date)
# Cross-rate: to_rate / from_rate
if from_rate == Decimal("0"):
raise DomainException(
code="INVALID_RATE",
message=f"Invalid zero rate for currency {from_currency_id}"
)
return to_rate / from_rate
async def convert_amount(
self,
tenant_id: str,
amount: Decimal,
from_currency_id: str,
to_currency_id: str,
target_date: Optional[date] = None,
round_result: bool = True
) -> Decimal:
"""
Convierte un monto entre dos monedas.
Args:
tenant_id: ID del tenant
amount: Monto a convertir
from_currency_id: Moneda origen
to_currency_id: Moneda destino
target_date: Fecha para la tasa
round_result: Si redondear según decimales de moneda destino
Returns:
Monto convertido
"""
if from_currency_id == to_currency_id:
return amount
rate = await self.get_rate_between(
tenant_id, from_currency_id, to_currency_id, target_date
)
converted = amount * rate
if round_result:
# Obtener decimales de moneda destino
currency_config = await self._get_currency_config(tenant_id, to_currency_id)
decimal_places = currency_config.decimal_places if currency_config else 2
converted = round(converted, decimal_places)
return converted
async def set_rate_manual(
self,
tenant_id: str,
currency_id: str,
rate: Decimal,
effective_date: date,
user_id: str,
notes: Optional[str] = None
) -> "CurrencyRate":
"""Establece una tasa manualmente."""
# Validar que no sea la moneda base
settings = await self._settings_repo.get_by_tenant(tenant_id)
if settings and settings.base_currency_id == currency_id:
raise DomainException(
code="CANNOT_SET_BASE_RATE",
message="Cannot set rate for base currency"
)
# Verificar cambio significativo
current_rate = await self._rate_repo.find_effective_rate(
tenant_id, currency_id, effective_date
)
if current_rate:
change_percent = abs((rate - current_rate.rate) / current_rate.rate * 100)
if change_percent > settings.max_rate_change_percent:
# Crear alerta
await self._alert_service.create_alert(
tenant_id=tenant_id,
currency_id=currency_id,
alert_type="manual_override",
previous_rate=current_rate.rate,
new_rate=rate,
change_percent=change_percent,
requires_approval=change_percent > settings.require_approval_above
)
# Crear o actualizar tasa
new_rate = await self._rate_repo.upsert(
tenant_id=tenant_id,
currency_id=currency_id,
effective_date=effective_date,
rate=rate,
source=RateSourceType.MANUAL,
created_by=user_id
)
# Invalidar cache
await self._invalidate_rate_cache(tenant_id, currency_id, effective_date)
logger.info(
"Manual rate set",
tenant_id=tenant_id,
currency_id=currency_id,
rate=str(rate),
date=str(effective_date),
user_id=user_id
)
return new_rate
async def get_rate_history(
self,
tenant_id: str,
currency_id: str,
date_from: date,
date_to: date
) -> List["CurrencyRate"]:
"""Obtiene historial de tasas para un período."""
return await self._rate_repo.find_by_date_range(
tenant_id=tenant_id,
currency_id=currency_id,
date_from=date_from,
date_to=date_to
)
def _calculate_ttl_to_midnight(self) -> int:
"""Calcula segundos hasta medianoche."""
now = datetime.now()
midnight = datetime.combine(now.date() + timedelta(days=1), time.min)
return int((midnight - now).total_seconds())
async def _get_currency_config(
self,
tenant_id: str,
currency_id: str
) -> Optional["CurrencyConfig"]:
"""Obtiene configuración de moneda para tenant."""
# Implementación del repositorio
pass
async def _invalidate_rate_cache(
self,
tenant_id: str,
currency_id: str,
effective_date: date
):
"""Invalida cache de tasa."""
cache_key = f"rate:{tenant_id}:{currency_id}:{effective_date}"
await self._cache.delete(cache_key)
class CurrencyRateUpdateService:
"""Servicio para actualización automática de tasas."""
def __init__(
self,
provider_registry: "ProviderRegistry",
rate_service: CurrencyRateService,
settings_repository: "TenantCurrencySettingsRepository",
log_repository: "UpdateLogRepository",
alert_service: "CurrencyRateAlertService"
):
self._providers = provider_registry
self._rate_service = rate_service
self._settings_repo = settings_repository
self._log_repo = log_repository
self._alert_service = alert_service
async def update_rates_for_tenant(
self,
tenant_id: str,
currencies: Optional[List[str]] = None,
force: bool = False
) -> "UpdateResult":
"""
Actualiza tasas de cambio para un tenant.
Args:
tenant_id: ID del tenant
currencies: Lista de códigos ISO a actualizar (None = todas activas)
force: Forzar actualización aunque ya exista tasa del día
Returns:
Resultado de la actualización
"""
settings = await self._settings_repo.get_by_tenant(tenant_id)
if not settings:
raise DomainException(
code="SETTINGS_NOT_FOUND",
message=f"Currency settings not found for tenant {tenant_id}"
)
# Obtener proveedor
provider = self._providers.get(settings.preferred_provider_id)
if not provider:
raise DomainException(
code="PROVIDER_NOT_FOUND",
message="No currency rate provider configured"
)
# Crear log de inicio
log = await self._log_repo.create(
tenant_id=tenant_id,
provider_id=provider.id,
status="in_progress"
)
start_time = datetime.utcnow()
result = UpdateResult()
try:
# Obtener tasas del proveedor
rates_data = await provider.fetch_rates(
currencies=currencies,
credentials=settings.api_credentials
)
result.currencies_requested = len(rates_data)
# Procesar cada tasa
for currency_code, rate_value in rates_data.items():
try:
await self._process_rate(
tenant_id=tenant_id,
currency_code=currency_code,
rate_value=rate_value,
provider_id=provider.id,
settings=settings,
force=force
)
result.currencies_updated += 1
except Exception as e:
result.currencies_failed += 1
result.errors.append(f"{currency_code}: {str(e)}")
# Determinar estado final
if result.currencies_failed == 0:
result.status = "success"
elif result.currencies_updated > 0:
result.status = "partial_success"
else:
result.status = "failed"
except Exception as e:
result.status = "failed"
result.error_message = str(e)
logger.exception("Rate update failed", tenant_id=tenant_id)
# Intentar con proveedor de respaldo
if settings.fallback_provider_id:
try:
result = await self._try_fallback_provider(
tenant_id, settings, currencies
)
except Exception as fallback_error:
logger.exception("Fallback provider also failed")
# Actualizar log
end_time = datetime.utcnow()
await self._log_repo.update(
log_id=log.id,
status=result.status,
completed_at=end_time,
duration_ms=int((end_time - start_time).total_seconds() * 1000),
currencies_requested=result.currencies_requested,
currencies_updated=result.currencies_updated,
currencies_failed=result.currencies_failed,
error_message=result.error_message
)
# Programar siguiente actualización
await self._schedule_next_update(tenant_id, settings)
return result
async def _process_rate(
self,
tenant_id: str,
currency_code: str,
rate_value: Decimal,
provider_id: str,
settings: "TenantCurrencySettings",
force: bool
):
"""Procesa y guarda una tasa individual."""
today = date.today()
# Verificar si ya existe tasa del día
if not force:
existing = await self._rate_service._rate_repo.find_by_date(
tenant_id, currency_code, today
)
if existing:
logger.debug(f"Rate for {currency_code} already exists for {today}")
return
# Obtener tasa anterior para validación
previous = await self._rate_service._rate_repo.find_effective_rate(
tenant_id, currency_code, today - timedelta(days=1)
)
# Validar cambio
if previous:
change_percent = abs((rate_value - previous.rate) / previous.rate * 100)
if change_percent > settings.max_rate_change_percent:
# Crear alerta pero continuar guardando
await self._alert_service.create_alert(
tenant_id=tenant_id,
currency_code=currency_code,
alert_type="large_change",
previous_rate=previous.rate,
new_rate=rate_value,
change_percent=change_percent,
requires_approval=change_percent > settings.require_approval_above
)
# Guardar tasa
await self._rate_service._rate_repo.upsert(
tenant_id=tenant_id,
currency_code=currency_code,
effective_date=today,
rate=rate_value,
source=RateSourceType.AUTOMATIC,
provider_id=provider_id
)
async def _schedule_next_update(
self,
tenant_id: str,
settings: "TenantCurrencySettings"
):
"""Calcula y guarda la próxima fecha de actualización."""
now = datetime.utcnow()
if settings.update_interval == "hourly":
next_update = now + timedelta(hours=1)
elif settings.update_interval == "daily":
# Próximo día a la hora configurada
next_date = now.date() + timedelta(days=1)
next_update = datetime.combine(next_date, settings.update_time)
elif settings.update_interval == "weekly":
# Próximo lunes
days_until_monday = (7 - now.weekday()) % 7 or 7
next_date = now.date() + timedelta(days=days_until_monday)
next_update = datetime.combine(next_date, settings.update_time)
elif settings.update_interval == "monthly":
# Primer día del próximo mes
if now.month == 12:
next_date = date(now.year + 1, 1, 1)
else:
next_date = date(now.year, now.month + 1, 1)
next_update = datetime.combine(next_date, settings.update_time)
else:
next_update = None
await self._settings_repo.update_next_scheduled(tenant_id, next_update)
class UpdateResult:
"""Resultado de actualización de tasas."""
def __init__(self):
self.status: str = "pending"
self.currencies_requested: int = 0
self.currencies_updated: int = 0
self.currencies_failed: int = 0
self.errors: List[str] = []
self.error_message: Optional[str] = None
3.3 Proveedores de Tasas
# app/modules/multi_currency/infrastructure/providers/base.py
from abc import ABC, abstractmethod
from decimal import Decimal
from typing import Dict, List, Optional
from datetime import date
class CurrencyRateProvider(ABC):
"""Clase base para proveedores de tasas de cambio."""
@property
@abstractmethod
def code(self) -> str:
"""Código único del proveedor."""
pass
@property
@abstractmethod
def name(self) -> str:
"""Nombre descriptivo."""
pass
@property
@abstractmethod
def base_currency(self) -> str:
"""Moneda base del proveedor (ISO 4217)."""
pass
@property
@abstractmethod
def supported_currencies(self) -> List[str]:
"""Lista de monedas soportadas."""
pass
@abstractmethod
async def fetch_rates(
self,
currencies: Optional[List[str]] = None,
target_date: Optional[date] = None,
credentials: Optional[Dict] = None
) -> Dict[str, Decimal]:
"""
Obtiene tasas de cambio del proveedor.
Args:
currencies: Monedas específicas a obtener (None = todas)
target_date: Fecha para tasas históricas (None = hoy)
credentials: Credenciales de API si requeridas
Returns:
Dict con código de moneda -> tasa relativa a base_currency
"""
pass
# app/modules/multi_currency/infrastructure/providers/ecb_provider.py
import httpx
from xml.etree import ElementTree as ET
from decimal import Decimal
from typing import Dict, List, Optional
from datetime import date
from .base import CurrencyRateProvider
class ECBProvider(CurrencyRateProvider):
"""
Proveedor de tasas del Banco Central Europeo.
- Base: EUR
- Actualización: Diaria ~16:00 CET
- Sin autenticación requerida
- ~32 monedas soportadas
"""
DAILY_URL = "https://www.ecb.europa.eu/stats/eurofxref/eurofxref-daily.xml"
HISTORY_URL = "https://www.ecb.europa.eu/stats/eurofxref/eurofxref-hist.xml"
# Namespace del XML del ECB
NS = {'gesmes': 'http://www.gesmes.org/xml/2002-08-01',
'eurofxref': 'http://www.ecb.int/vocabulary/2002-08-01/eurofxref'}
@property
def code(self) -> str:
return "ecb"
@property
def name(self) -> str:
return "European Central Bank"
@property
def base_currency(self) -> str:
return "EUR"
@property
def supported_currencies(self) -> List[str]:
return [
"USD", "JPY", "BGN", "CZK", "DKK", "GBP", "HUF", "PLN",
"RON", "SEK", "CHF", "ISK", "NOK", "TRY", "AUD", "BRL",
"CAD", "CNY", "HKD", "IDR", "ILS", "INR", "KRW", "MXN",
"MYR", "NZD", "PHP", "SGD", "THB", "ZAR"
]
async def fetch_rates(
self,
currencies: Optional[List[str]] = None,
target_date: Optional[date] = None,
credentials: Optional[Dict] = None
) -> Dict[str, Decimal]:
"""Obtiene tasas del BCE."""
url = self.DAILY_URL if not target_date else self.HISTORY_URL
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.get(url)
response.raise_for_status()
rates = self._parse_xml(response.content, target_date)
# Filtrar monedas si se especificaron
if currencies:
rates = {k: v for k, v in rates.items() if k in currencies}
# EUR siempre es 1.0
rates["EUR"] = Decimal("1.0")
return rates
def _parse_xml(
self,
content: bytes,
target_date: Optional[date] = None
) -> Dict[str, Decimal]:
"""Parsea respuesta XML del BCE."""
root = ET.fromstring(content)
rates = {}
# Buscar el Cube con los datos
for time_cube in root.findall('.//eurofxref:Cube[@time]', self.NS):
cube_date = time_cube.get('time')
# Si buscamos fecha específica, verificar
if target_date and cube_date != str(target_date):
continue
for rate_cube in time_cube.findall('eurofxref:Cube', self.NS):
currency = rate_cube.get('currency')
rate = rate_cube.get('rate')
if currency and rate:
rates[currency] = Decimal(rate)
# Si encontramos datos (o es el más reciente), salir
if rates:
break
return rates
# app/modules/multi_currency/infrastructure/providers/banxico_provider.py
class BanxicoProvider(CurrencyRateProvider):
"""
Proveedor de tasas del Banco de México.
- Base: MXN
- Requiere token de API
- Series disponibles: USD/MXN, EUR/MXN, etc.
"""
BASE_URL = "https://www.banxico.org.mx/SieAPIRest/service/v1/series"
# Series del SIE para diferentes monedas
SERIES = {
"USD": "SF43718", # Tipo de cambio pesos por dólar E.U.A.
"EUR": "SF46410", # Tipo de cambio pesos por euro
"GBP": "SF46407", # Tipo de cambio pesos por libra esterlina
"JPY": "SF46406", # Tipo de cambio pesos por yen japonés
"CAD": "SF60632", # Tipo de cambio pesos por dólar canadiense
}
@property
def code(self) -> str:
return "banxico"
@property
def name(self) -> str:
return "Banco de México"
@property
def base_currency(self) -> str:
return "MXN"
@property
def supported_currencies(self) -> List[str]:
return list(self.SERIES.keys())
async def fetch_rates(
self,
currencies: Optional[List[str]] = None,
target_date: Optional[date] = None,
credentials: Optional[Dict] = None
) -> Dict[str, Decimal]:
"""Obtiene tasas de Banxico."""
if not credentials or 'token' not in credentials:
raise ValueError("Banxico API token required")
token = credentials['token']
currencies = currencies or list(self.SERIES.keys())
# Construir lista de series a consultar
series_ids = [self.SERIES[c] for c in currencies if c in self.SERIES]
series_param = ",".join(series_ids)
url = f"{self.BASE_URL}/{series_param}/datos/oportuno"
headers = {"Bmx-Token": token}
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.get(url, headers=headers)
response.raise_for_status()
data = response.json()
rates = self._parse_response(data, currencies)
# MXN siempre es 1.0
rates["MXN"] = Decimal("1.0")
return rates
def _parse_response(
self,
data: Dict,
currencies: List[str]
) -> Dict[str, Decimal]:
"""Parsea respuesta JSON de Banxico."""
rates = {}
series_list = data.get("bmx", {}).get("series", [])
for series in series_list:
series_id = series.get("idSerie")
# Encontrar moneda correspondiente
currency = None
for curr, sid in self.SERIES.items():
if sid == series_id:
currency = curr
break
if currency and currency in currencies:
datos = series.get("datos", [])
if datos:
# Tomar el dato más reciente
rate_str = datos[0].get("dato", "").replace(",", "")
if rate_str and rate_str != "N/E":
rates[currency] = Decimal(rate_str)
return rates
# app/modules/multi_currency/infrastructure/providers/bcb_provider.py
class BCBProvider(CurrencyRateProvider):
"""
Proveedor de tasas del Banco Central do Brasil.
- Base: BRL
- API PTAX (tasa oficial)
- Sin autenticación
"""
BASE_URL = "https://olinda.bcb.gov.br/olinda/servico/PTAX/versao/v1/odata"
@property
def code(self) -> str:
return "bcb"
@property
def name(self) -> str:
return "Banco Central do Brasil"
@property
def base_currency(self) -> str:
return "BRL"
@property
def supported_currencies(self) -> List[str]:
return ["USD", "EUR", "GBP", "JPY", "CHF", "AUD", "CAD"]
async def fetch_rates(
self,
currencies: Optional[List[str]] = None,
target_date: Optional[date] = None,
credentials: Optional[Dict] = None
) -> Dict[str, Decimal]:
"""Obtiene tasas PTAX del BCB."""
target_date = target_date or date.today()
date_str = target_date.strftime("%m-%d-%Y")
rates = {}
currencies = currencies or self.supported_currencies
async with httpx.AsyncClient(timeout=30.0) as client:
for currency in currencies:
if currency not in self.supported_currencies:
continue
url = (
f"{self.BASE_URL}/CotacaoMoedaDia(moeda=@moeda,dataCotacao=@data)"
f"?@moeda='{currency}'&@data='{date_str}'&$format=json"
)
try:
response = await client.get(url)
response.raise_for_status()
data = response.json()
values = data.get("value", [])
if values:
# Usar cotação de compra (buying rate)
rate = values[-1].get("cotacaoCompra")
if rate:
rates[currency] = Decimal(str(rate))
except Exception as e:
# Log error but continue with other currencies
pass
# BRL siempre es 1.0
rates["BRL"] = Decimal("1.0")
return rates
# app/modules/multi_currency/infrastructure/providers/registry.py
class ProviderRegistry:
"""Registro de proveedores de tasas de cambio."""
def __init__(self):
self._providers: Dict[str, CurrencyRateProvider] = {}
self._register_defaults()
def _register_defaults(self):
"""Registra proveedores predeterminados."""
self.register(ECBProvider())
self.register(BanxicoProvider())
self.register(BCBProvider())
def register(self, provider: CurrencyRateProvider):
"""Registra un proveedor."""
self._providers[provider.code] = provider
def get(self, code: str) -> Optional[CurrencyRateProvider]:
"""Obtiene un proveedor por código."""
return self._providers.get(code)
def get_by_id(self, provider_id: str) -> Optional[CurrencyRateProvider]:
"""Obtiene un proveedor por ID de base de datos."""
# Buscar en DB y retornar proveedor correspondiente
pass
def list_all(self) -> List[CurrencyRateProvider]:
"""Lista todos los proveedores registrados."""
return list(self._providers.values())
3.4 Scheduler (Cron Jobs)
# app/modules/multi_currency/infrastructure/scheduler/rate_update_scheduler.py
from datetime import datetime, timedelta
from typing import List, Optional
import asyncio
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger
from app.core.logging import get_logger
from app.modules.multi_currency.domain.services.currency_rate_service import (
CurrencyRateUpdateService
)
logger = get_logger(__name__)
class RateUpdateScheduler:
"""
Programador de actualizaciones de tasas de cambio.
Ejecuta actualizaciones automáticas según configuración de cada tenant.
"""
def __init__(
self,
update_service: CurrencyRateUpdateService,
tenant_service: "TenantService"
):
self._update_service = update_service
self._tenant_service = tenant_service
self._scheduler = AsyncIOScheduler()
self._running = False
async def start(self):
"""Inicia el scheduler."""
if self._running:
return
# Job maestro que verifica tenants pendientes cada 5 minutos
self._scheduler.add_job(
self._check_pending_updates,
trigger=IntervalTrigger(minutes=5),
id="rate_update_checker",
name="Currency Rate Update Checker",
replace_existing=True
)
# Job de limpieza de logs antiguos (diario a las 3 AM)
self._scheduler.add_job(
self._cleanup_old_logs,
trigger=CronTrigger(hour=3, minute=0),
id="rate_log_cleanup",
name="Rate Update Log Cleanup",
replace_existing=True
)
self._scheduler.start()
self._running = True
logger.info("Rate update scheduler started")
async def stop(self):
"""Detiene el scheduler."""
if not self._running:
return
self._scheduler.shutdown(wait=True)
self._running = False
logger.info("Rate update scheduler stopped")
async def _check_pending_updates(self):
"""
Verifica y ejecuta actualizaciones pendientes.
Busca tenants cuya próxima actualización programada ha pasado
y ejecuta la actualización de tasas.
"""
try:
now = datetime.utcnow()
# Obtener tenants con actualización pendiente
pending_tenants = await self._tenant_service.get_tenants_pending_rate_update(
before=now
)
logger.info(
f"Found {len(pending_tenants)} tenants pending rate update"
)
# Procesar cada tenant (con límite de concurrencia)
semaphore = asyncio.Semaphore(5) # Máximo 5 concurrentes
async def update_with_limit(tenant_id: str):
async with semaphore:
await self._update_tenant_rates(tenant_id)
await asyncio.gather(*[
update_with_limit(t.id) for t in pending_tenants
])
except Exception as e:
logger.exception("Error checking pending rate updates")
async def _update_tenant_rates(self, tenant_id: str):
"""Ejecuta actualización de tasas para un tenant."""
try:
logger.info(f"Starting rate update for tenant {tenant_id}")
result = await self._update_service.update_rates_for_tenant(tenant_id)
logger.info(
f"Rate update completed for tenant {tenant_id}",
status=result.status,
updated=result.currencies_updated,
failed=result.currencies_failed
)
except Exception as e:
logger.exception(
f"Rate update failed for tenant {tenant_id}",
error=str(e)
)
async def _cleanup_old_logs(self):
"""Limpia logs de actualización antiguos."""
try:
# Obtener configuración de retención por tenant
tenants = await self._tenant_service.get_all_active()
for tenant in tenants:
settings = await self._update_service._settings_repo.get_by_tenant(
tenant.id
)
if settings:
cutoff_date = datetime.utcnow() - timedelta(
days=settings.keep_history_days
)
deleted = await self._update_service._log_repo.delete_before(
tenant_id=tenant.id,
before=cutoff_date
)
if deleted > 0:
logger.info(
f"Deleted {deleted} old rate update logs",
tenant_id=tenant.id
)
except Exception as e:
logger.exception("Error cleaning up old rate logs")
async def trigger_immediate_update(
self,
tenant_id: str,
currencies: Optional[List[str]] = None
):
"""Dispara actualización inmediata para un tenant."""
logger.info(
f"Triggering immediate rate update for tenant {tenant_id}",
currencies=currencies
)
return await self._update_service.update_rates_for_tenant(
tenant_id=tenant_id,
currencies=currencies,
force=True
)
3.5 API REST
# app/modules/multi_currency/api/v1/currency_rates.py
from datetime import date
from decimal import Decimal
from typing import List, Optional
from fastapi import APIRouter, Depends, Query, Path, HTTPException, status
from pydantic import BaseModel, Field
from app.core.auth import get_current_user, require_permissions
from app.core.pagination import PaginatedResponse, PaginationParams
from app.modules.multi_currency.domain.services.currency_rate_service import (
CurrencyRateService,
CurrencyRateUpdateService
)
router = APIRouter(prefix="/currency-rates", tags=["Currency Rates"])
# =====================================================
# SCHEMAS
# =====================================================
class CurrencyRateResponse(BaseModel):
"""Respuesta de tasa de cambio."""
currency_id: str
currency_code: str
currency_name: str
rate: Decimal
inverse_rate: Decimal
effective_date: date
source: str
provider_name: Optional[str] = None
class CurrencyRateHistoryResponse(BaseModel):
"""Respuesta de historial de tasas."""
currency_id: str
currency_code: str
rates: List[CurrencyRateResponse]
class ConversionRequest(BaseModel):
"""Request para conversión de moneda."""
amount: Decimal = Field(..., gt=0)
from_currency_id: str
to_currency_id: str
date: Optional[date] = None
class ConversionResponse(BaseModel):
"""Respuesta de conversión."""
original_amount: Decimal
from_currency: str
converted_amount: Decimal
to_currency: str
rate_used: Decimal
rate_date: date
class SetRateRequest(BaseModel):
"""Request para establecer tasa manual."""
currency_id: str
rate: Decimal = Field(..., gt=0)
effective_date: date
notes: Optional[str] = None
class UpdateRatesRequest(BaseModel):
"""Request para actualizar tasas."""
currencies: Optional[List[str]] = None # Códigos ISO
force: bool = False
class UpdateRatesResponse(BaseModel):
"""Respuesta de actualización de tasas."""
status: str
currencies_requested: int
currencies_updated: int
currencies_failed: int
errors: List[str] = []
class TenantCurrencySettingsRequest(BaseModel):
"""Request para configuración de moneda del tenant."""
base_currency_id: str
preferred_provider_code: str
fallback_provider_code: Optional[str] = None
update_interval: str = "daily"
update_time: str = "06:00:00"
update_timezone: str = "UTC"
max_rate_change_percent: Decimal = Decimal("10.00")
require_approval_above: Decimal = Decimal("20.00")
class TenantCurrencySettingsResponse(BaseModel):
"""Respuesta de configuración de moneda."""
base_currency_id: str
base_currency_code: str
preferred_provider: str
fallback_provider: Optional[str]
update_interval: str
update_time: str
update_timezone: str
next_scheduled_update: Optional[str]
max_rate_change_percent: Decimal
require_approval_above: Decimal
# =====================================================
# ENDPOINTS
# =====================================================
@router.get("/current", response_model=List[CurrencyRateResponse])
async def get_current_rates(
currencies: Optional[str] = Query(None, description="Códigos ISO separados por coma"),
rate_service: CurrencyRateService = Depends(),
current_user = Depends(get_current_user)
):
"""
Obtiene las tasas de cambio vigentes para el tenant.
- Sin parámetros: retorna todas las monedas activas
- Con `currencies`: retorna solo las monedas especificadas
"""
tenant_id = current_user.tenant_id
currency_list = currencies.split(",") if currencies else None
rates = await rate_service.get_current_rates(
tenant_id=tenant_id,
currency_codes=currency_list
)
return [CurrencyRateResponse(**r.dict()) for r in rates]
@router.get("/{currency_id}", response_model=CurrencyRateResponse)
async def get_rate(
currency_id: str = Path(..., description="ID de la moneda"),
date: Optional[date] = Query(None, description="Fecha para la tasa"),
rate_service: CurrencyRateService = Depends(),
current_user = Depends(get_current_user)
):
"""Obtiene la tasa de una moneda específica."""
try:
rate = await rate_service.get_rate(
tenant_id=current_user.tenant_id,
currency_id=currency_id,
target_date=date
)
return CurrencyRateResponse(**rate.dict())
except Exception as e:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=str(e)
)
@router.get("/{currency_id}/history", response_model=CurrencyRateHistoryResponse)
async def get_rate_history(
currency_id: str = Path(..., description="ID de la moneda"),
date_from: date = Query(..., description="Fecha inicio"),
date_to: date = Query(..., description="Fecha fin"),
rate_service: CurrencyRateService = Depends(),
current_user = Depends(get_current_user)
):
"""Obtiene el historial de tasas de una moneda."""
rates = await rate_service.get_rate_history(
tenant_id=current_user.tenant_id,
currency_id=currency_id,
date_from=date_from,
date_to=date_to
)
return CurrencyRateHistoryResponse(
currency_id=currency_id,
currency_code=rates[0].currency_code if rates else "",
rates=[CurrencyRateResponse(**r.dict()) for r in rates]
)
@router.post("/convert", response_model=ConversionResponse)
async def convert_currency(
request: ConversionRequest,
rate_service: CurrencyRateService = Depends(),
current_user = Depends(get_current_user)
):
"""Convierte un monto entre dos monedas."""
rate = await rate_service.get_rate_between(
tenant_id=current_user.tenant_id,
from_currency_id=request.from_currency_id,
to_currency_id=request.to_currency_id,
target_date=request.date
)
converted = await rate_service.convert_amount(
tenant_id=current_user.tenant_id,
amount=request.amount,
from_currency_id=request.from_currency_id,
to_currency_id=request.to_currency_id,
target_date=request.date
)
return ConversionResponse(
original_amount=request.amount,
from_currency=request.from_currency_id,
converted_amount=converted,
to_currency=request.to_currency_id,
rate_used=rate,
rate_date=request.date or date.today()
)
@router.post("/", response_model=CurrencyRateResponse)
@require_permissions(["currency_rates:write"])
async def set_rate_manual(
request: SetRateRequest,
rate_service: CurrencyRateService = Depends(),
current_user = Depends(get_current_user)
):
"""
Establece una tasa de cambio manualmente.
Requiere permiso `currency_rates:write`.
"""
rate = await rate_service.set_rate_manual(
tenant_id=current_user.tenant_id,
currency_id=request.currency_id,
rate=request.rate,
effective_date=request.effective_date,
user_id=current_user.id,
notes=request.notes
)
return CurrencyRateResponse(**rate.dict())
@router.post("/update", response_model=UpdateRatesResponse)
@require_permissions(["currency_rates:update"])
async def update_rates(
request: UpdateRatesRequest,
update_service: CurrencyRateUpdateService = Depends(),
current_user = Depends(get_current_user)
):
"""
Dispara actualización de tasas desde proveedor externo.
Requiere permiso `currency_rates:update`.
"""
result = await update_service.update_rates_for_tenant(
tenant_id=current_user.tenant_id,
currencies=request.currencies,
force=request.force
)
return UpdateRatesResponse(
status=result.status,
currencies_requested=result.currencies_requested,
currencies_updated=result.currencies_updated,
currencies_failed=result.currencies_failed,
errors=result.errors
)
# =====================================================
# CONFIGURACIÓN
# =====================================================
@router.get("/settings", response_model=TenantCurrencySettingsResponse)
async def get_settings(
settings_service = Depends(),
current_user = Depends(get_current_user)
):
"""Obtiene la configuración de monedas del tenant."""
settings = await settings_service.get_by_tenant(current_user.tenant_id)
if not settings:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Currency settings not configured"
)
return TenantCurrencySettingsResponse(**settings.dict())
@router.put("/settings", response_model=TenantCurrencySettingsResponse)
@require_permissions(["currency_settings:write"])
async def update_settings(
request: TenantCurrencySettingsRequest,
settings_service = Depends(),
current_user = Depends(get_current_user)
):
"""
Actualiza la configuración de monedas del tenant.
Requiere permiso `currency_settings:write`.
"""
settings = await settings_service.upsert(
tenant_id=current_user.tenant_id,
**request.dict()
)
return TenantCurrencySettingsResponse(**settings.dict())
# =====================================================
# PROVEEDORES
# =====================================================
class ProviderInfo(BaseModel):
code: str
name: str
base_currency: str
supported_currencies: List[str]
@router.get("/providers", response_model=List[ProviderInfo])
async def list_providers(
provider_registry = Depends()
):
"""Lista los proveedores de tasas disponibles."""
providers = provider_registry.list_all()
return [
ProviderInfo(
code=p.code,
name=p.name,
base_currency=p.base_currency,
supported_currencies=p.supported_currencies
)
for p in providers
]
4. Reglas de Negocio
4.1 Gestión de Tasas
| Regla |
Descripción |
| RN-TCR-001 |
Solo una tasa por moneda por día por tenant |
| RN-TCR-002 |
La moneda base del tenant siempre tiene tasa 1.0 |
| RN-TCR-003 |
Cross-rates se calculan automáticamente vía moneda base |
| RN-TCR-004 |
Tasas con más de X% de cambio generan alertas |
| RN-TCR-005 |
Cambios mayores a umbral requieren aprobación manual |
| RN-TCR-006 |
Tasas históricas son inmutables después de cierre de período |
4.2 Actualización Automática
| Regla |
Descripción |
| RN-TCR-007 |
Si falla proveedor principal, usar proveedor de respaldo |
| RN-TCR-008 |
Máximo 3 reintentos con backoff exponencial |
| RN-TCR-009 |
Si falla todo, mantener última tasa conocida y alertar |
| RN-TCR-010 |
No sobrescribir tasas manuales del mismo día |
| RN-TCR-011 |
Rate limiting según configuración del proveedor |
4.3 Conversión
| Regla |
Descripción |
| RN-TCR-012 |
Usar tasa del día de transacción, no del día actual |
| RN-TCR-013 |
Si no hay tasa del día exacto, usar la más reciente anterior |
| RN-TCR-014 |
Redondear según decimales de moneda destino |
| RN-TCR-015 |
Registrar tasa usada en cada transacción para auditoría |
5. Pruebas
5.1 Casos de Prueba Unitarios
# tests/unit/multi_currency/test_currency_rate_service.py
import pytest
from decimal import Decimal
from datetime import date, timedelta
from unittest.mock import AsyncMock, MagicMock
from app.modules.multi_currency.domain.services.currency_rate_service import (
CurrencyRateService,
RateSourceType
)
class TestCurrencyRateService:
"""Tests para CurrencyRateService."""
@pytest.fixture
def rate_repository(self):
return AsyncMock()
@pytest.fixture
def provider_repository(self):
return AsyncMock()
@pytest.fixture
def settings_repository(self):
return AsyncMock()
@pytest.fixture
def alert_service(self):
return AsyncMock()
@pytest.fixture
def cache_service(self):
return AsyncMock()
@pytest.fixture
def service(
self,
rate_repository,
provider_repository,
settings_repository,
alert_service,
cache_service
):
return CurrencyRateService(
rate_repository=rate_repository,
provider_repository=provider_repository,
settings_repository=settings_repository,
alert_service=alert_service,
cache_service=cache_service
)
async def test_get_rate_from_cache(self, service, cache_service):
"""Debe retornar tasa desde cache si existe."""
cache_service.get.return_value = "1.2345"
rate = await service.get_rate(
tenant_id="tenant-1",
currency_id="currency-usd",
target_date=date.today()
)
assert rate == Decimal("1.2345")
cache_service.get.assert_called_once()
async def test_get_rate_from_database(
self,
service,
cache_service,
rate_repository
):
"""Debe obtener tasa de DB si no está en cache."""
cache_service.get.return_value = None
mock_rate = MagicMock()
mock_rate.rate = Decimal("1.5678")
rate_repository.find_effective_rate.return_value = mock_rate
rate = await service.get_rate(
tenant_id="tenant-1",
currency_id="currency-eur",
target_date=date.today()
)
assert rate == Decimal("1.5678")
rate_repository.find_effective_rate.assert_called_once()
cache_service.set.assert_called_once()
async def test_get_rate_base_currency_returns_one(
self,
service,
cache_service,
rate_repository,
settings_repository
):
"""Debe retornar 1.0 para moneda base."""
cache_service.get.return_value = None
rate_repository.find_effective_rate.return_value = None
mock_settings = MagicMock()
mock_settings.base_currency_id = "currency-mxn"
settings_repository.get_by_tenant.return_value = mock_settings
rate = await service.get_rate(
tenant_id="tenant-1",
currency_id="currency-mxn",
target_date=date.today()
)
assert rate == Decimal("1.0")
async def test_convert_amount(self, service):
"""Debe convertir monto correctamente."""
service.get_rate_between = AsyncMock(return_value=Decimal("20.5"))
service._get_currency_config = AsyncMock(return_value=MagicMock(decimal_places=2))
converted = await service.convert_amount(
tenant_id="tenant-1",
amount=Decimal("100"),
from_currency_id="currency-usd",
to_currency_id="currency-mxn"
)
assert converted == Decimal("2050.00")
async def test_convert_same_currency(self, service):
"""Debe retornar mismo monto si monedas son iguales."""
converted = await service.convert_amount(
tenant_id="tenant-1",
amount=Decimal("100"),
from_currency_id="currency-usd",
to_currency_id="currency-usd"
)
assert converted == Decimal("100")
async def test_set_rate_manual_creates_alert_on_large_change(
self,
service,
rate_repository,
settings_repository,
alert_service
):
"""Debe crear alerta si cambio es mayor al umbral."""
# Configuración con umbral de 5%
mock_settings = MagicMock()
mock_settings.base_currency_id = "currency-mxn"
mock_settings.max_rate_change_percent = Decimal("5.0")
mock_settings.require_approval_above = Decimal("10.0")
settings_repository.get_by_tenant.return_value = mock_settings
# Tasa actual: 20.0
mock_current = MagicMock()
mock_current.rate = Decimal("20.0")
rate_repository.find_effective_rate.return_value = mock_current
# Nueva tasa: 22.0 (10% de cambio)
rate_repository.upsert.return_value = MagicMock()
await service.set_rate_manual(
tenant_id="tenant-1",
currency_id="currency-usd",
rate=Decimal("22.0"),
effective_date=date.today(),
user_id="user-1"
)
alert_service.create_alert.assert_called_once()
class TestCrossRate:
"""Tests para cálculo de cross-rates."""
async def test_cross_rate_calculation(self):
"""Debe calcular cross-rate correctamente."""
# USD/MXN = 20.0, EUR/MXN = 22.0
# EUR/USD = 22.0 / 20.0 = 1.1
service = CurrencyRateService(
rate_repository=AsyncMock(),
provider_repository=AsyncMock(),
settings_repository=AsyncMock(),
alert_service=AsyncMock(),
cache_service=AsyncMock()
)
async def mock_get_rate(tenant_id, currency_id, target_date):
rates = {
"currency-usd": Decimal("20.0"),
"currency-eur": Decimal("22.0"),
"currency-mxn": Decimal("1.0")
}
return rates.get(currency_id, Decimal("1.0"))
service.get_rate = mock_get_rate
cross_rate = await service.get_rate_between(
tenant_id="tenant-1",
from_currency_id="currency-usd",
to_currency_id="currency-eur",
target_date=date.today()
)
assert cross_rate == Decimal("1.1")
5.2 Tests de Proveedores
# tests/unit/multi_currency/test_providers.py
import pytest
from decimal import Decimal
from unittest.mock import AsyncMock, patch, MagicMock
import httpx
from app.modules.multi_currency.infrastructure.providers.ecb_provider import (
ECBProvider
)
from app.modules.multi_currency.infrastructure.providers.banxico_provider import (
BanxicoProvider
)
class TestECBProvider:
"""Tests para proveedor ECB."""
@pytest.fixture
def provider(self):
return ECBProvider()
@pytest.fixture
def sample_ecb_response(self):
return b"""<?xml version="1.0" encoding="UTF-8"?>
<gesmes:Envelope xmlns:gesmes="http://www.gesmes.org/xml/2002-08-01"
xmlns="http://www.ecb.int/vocabulary/2002-08-01/eurofxref">
<Cube>
<Cube time="2024-01-15">
<Cube currency="USD" rate="1.0876"/>
<Cube currency="JPY" rate="160.88"/>
<Cube currency="GBP" rate="0.85680"/>
<Cube currency="MXN" rate="18.5432"/>
</Cube>
</Cube>
</gesmes:Envelope>
"""
async def test_fetch_rates_success(self, provider, sample_ecb_response):
"""Debe parsear respuesta del ECB correctamente."""
mock_response = MagicMock()
mock_response.content = sample_ecb_response
mock_response.raise_for_status = MagicMock()
with patch.object(httpx.AsyncClient, 'get', return_value=mock_response):
rates = await provider.fetch_rates()
assert "USD" in rates
assert rates["USD"] == Decimal("1.0876")
assert rates["EUR"] == Decimal("1.0")
assert "MXN" in rates
async def test_fetch_specific_currencies(self, provider, sample_ecb_response):
"""Debe filtrar monedas solicitadas."""
mock_response = MagicMock()
mock_response.content = sample_ecb_response
mock_response.raise_for_status = MagicMock()
with patch.object(httpx.AsyncClient, 'get', return_value=mock_response):
rates = await provider.fetch_rates(currencies=["USD", "GBP"])
assert "USD" in rates
assert "GBP" in rates
assert "JPY" not in rates
assert "EUR" in rates # EUR siempre incluido
class TestBanxicoProvider:
"""Tests para proveedor Banxico."""
@pytest.fixture
def provider(self):
return BanxicoProvider()
@pytest.fixture
def sample_banxico_response(self):
return {
"bmx": {
"series": [
{
"idSerie": "SF43718",
"datos": [
{"fecha": "15/01/2024", "dato": "17.1234"}
]
}
]
}
}
async def test_fetch_rates_success(self, provider, sample_banxico_response):
"""Debe parsear respuesta de Banxico correctamente."""
mock_response = MagicMock()
mock_response.json.return_value = sample_banxico_response
mock_response.raise_for_status = MagicMock()
with patch.object(httpx.AsyncClient, 'get', return_value=mock_response):
rates = await provider.fetch_rates(
credentials={"token": "test-token"}
)
assert "USD" in rates
assert rates["USD"] == Decimal("17.1234")
assert rates["MXN"] == Decimal("1.0")
async def test_fetch_rates_requires_token(self, provider):
"""Debe fallar sin token de API."""
with pytest.raises(ValueError, match="token required"):
await provider.fetch_rates()
5.3 Tests de Integración
# tests/integration/multi_currency/test_rate_update_flow.py
import pytest
from datetime import date
from decimal import Decimal
from app.modules.multi_currency.domain.services.currency_rate_service import (
CurrencyRateService,
CurrencyRateUpdateService
)
@pytest.mark.integration
class TestRateUpdateFlow:
"""Tests de integración para flujo de actualización."""
async def test_full_update_flow(
self,
db_session,
test_tenant,
rate_service: CurrencyRateService,
update_service: CurrencyRateUpdateService
):
"""Test del flujo completo de actualización de tasas."""
# 1. Configurar tenant con proveedor ECB
await self._setup_tenant_settings(
db_session,
test_tenant.id,
provider="ecb"
)
# 2. Ejecutar actualización
result = await update_service.update_rates_for_tenant(
tenant_id=test_tenant.id,
currencies=["USD", "GBP", "JPY"]
)
# 3. Verificar resultado
assert result.status in ["success", "partial_success"]
assert result.currencies_updated > 0
# 4. Verificar tasas guardadas
usd_rate = await rate_service.get_rate(
tenant_id=test_tenant.id,
currency_id="currency-usd",
target_date=date.today()
)
assert usd_rate > Decimal("0")
async def test_conversion_uses_correct_rate(
self,
db_session,
test_tenant,
rate_service: CurrencyRateService
):
"""Test que conversión usa tasa correcta."""
# Setup: crear tasas conocidas
await rate_service.set_rate_manual(
tenant_id=test_tenant.id,
currency_id="currency-usd",
rate=Decimal("17.50"),
effective_date=date.today(),
user_id="test-user"
)
# Convertir 100 USD a MXN
converted = await rate_service.convert_amount(
tenant_id=test_tenant.id,
amount=Decimal("100"),
from_currency_id="currency-usd",
to_currency_id="currency-mxn" # MXN es base
)
# MXN base = 1.0, USD = 17.50
# 100 USD * (1.0 / 17.50) = ~5.71 MXN (invertido)
# O: 100 USD en términos de MXN = 100 * 17.50 = 1750 MXN
assert converted == Decimal("1750.00")
6. Configuración y Despliegue
6.1 Variables de Entorno
# .env.example
# Configuración general de tasas
CURRENCY_RATE_CACHE_TTL=86400
CURRENCY_RATE_UPDATE_BATCH_SIZE=10
CURRENCY_RATE_MAX_RETRIES=3
CURRENCY_RATE_RETRY_DELAY_SECONDS=60
# Proveedor ECB
ECB_API_URL=https://www.ecb.europa.eu/stats/eurofxref/eurofxref-daily.xml
ECB_TIMEOUT_SECONDS=30
# Proveedor Banxico (requiere token por tenant)
BANXICO_API_URL=https://www.banxico.org.mx/SieAPIRest/service/v1/series
# Proveedor BCB
BCB_API_URL=https://olinda.bcb.gov.br/olinda/servico/PTAX/versao/v1/odata
# Alertas
RATE_CHANGE_ALERT_THRESHOLD_PERCENT=10
RATE_CHANGE_APPROVAL_THRESHOLD_PERCENT=20
# Scheduler
RATE_UPDATE_CHECK_INTERVAL_MINUTES=5
RATE_LOG_RETENTION_DAYS=365
6.2 Migraciones
-- migrations/xxx_create_currency_rates.sql
-- Extensión para parent_path si no existe
CREATE EXTENSION IF NOT EXISTS ltree;
-- Ejecutar DDL del modelo de datos (sección 3.1)
-- ...
-- Índices adicionales para rendimiento
CREATE INDEX CONCURRENTLY idx_currency_rates_tenant_date
ON currency_rates(tenant_id, effective_date DESC);
CREATE INDEX CONCURRENTLY idx_rate_alerts_pending
ON currency_rate_alerts(tenant_id, status)
WHERE status = 'pending';
-- Datos iniciales de proveedores
INSERT INTO currency_rate_providers (code, name, provider_type, base_currency_id, api_url, response_format)
SELECT 'ecb', 'European Central Bank', 'ecb', c.id,
'https://www.ecb.europa.eu/stats/eurofxref/eurofxref-daily.xml', 'xml'
FROM currencies c WHERE c.iso_code = 'EUR';
7. Métricas y Monitoreo
7.1 Métricas Clave
| Métrica |
Tipo |
Descripción |
currency_rate_updates_total |
Counter |
Total de actualizaciones ejecutadas |
currency_rate_updates_success |
Counter |
Actualizaciones exitosas |
currency_rate_updates_failed |
Counter |
Actualizaciones fallidas |
currency_rate_update_duration_ms |
Histogram |
Duración de actualización |
currency_rate_conversions_total |
Counter |
Total de conversiones realizadas |
currency_rate_cache_hits |
Counter |
Cache hits de tasas |
currency_rate_cache_misses |
Counter |
Cache misses de tasas |
currency_rate_alerts_total |
Counter |
Total de alertas generadas |
7.2 Alertas de Monitoreo
# alerts/currency_rates.yaml
groups:
- name: currency_rates
rules:
- alert: CurrencyRateUpdateFailed
expr: increase(currency_rate_updates_failed[1h]) > 3
for: 5m
labels:
severity: warning
annotations:
summary: "Currency rate updates failing"
description: "More than 3 rate update failures in the last hour"
- alert: CurrencyRateStale
expr: currency_rate_age_hours > 48
for: 1h
labels:
severity: critical
annotations:
summary: "Currency rates are stale"
description: "Currency rates haven't been updated in 48+ hours"
- alert: CurrencyRateProviderDown
expr: currency_rate_provider_consecutive_failures > 5
for: 10m
labels:
severity: warning
annotations:
summary: "Currency rate provider failing"
description: "Provider {{ $labels.provider }} has failed 5+ consecutive times"
8. Dependencias
8.1 Dependencias de Sistema
| Componente |
Versión |
Propósito |
| PostgreSQL |
15+ |
Base de datos |
| Redis |
7+ |
Cache de tasas |
| APScheduler |
3.10+ |
Programación de jobs |
8.2 Dependencias de Módulos
| Módulo |
Tipo |
Descripción |
base |
Requerido |
Currencies, tenants |
accounting |
Opcional |
Integración con asientos |
notifications |
Opcional |
Alertas de cambios |
9. Anexos
9.1 Proveedores Disponibles
| Proveedor |
Código |
Moneda Base |
Autenticación |
Límite |
| BCE |
ecb |
EUR |
No |
Ilimitado |
| Banxico |
banxico |
MXN |
Token |
1000/día |
| BCB |
bcb |
BRL |
No |
Ilimitado |
| BCRP |
bcrp |
PEN |
No |
Ilimitado |
| BOC |
boc |
CAD |
No |
Ilimitado |
| Fixer.io |
fixer |
EUR |
API Key |
Según plan |
| Open Exchange |
openexchange |
USD |
App ID |
Según plan |
9.2 Códigos de Error
| Código |
Descripción |
RATE_NOT_FOUND |
No se encontró tasa para fecha/moneda |
INVALID_RATE |
Tasa con valor inválido (0 o negativo) |
CANNOT_SET_BASE_RATE |
Intento de modificar tasa de moneda base |
PROVIDER_NOT_FOUND |
Proveedor no configurado |
PROVIDER_ERROR |
Error de comunicación con proveedor |
RATE_LIMIT_EXCEEDED |
Límite de API excedido |
SETTINGS_NOT_FOUND |
Configuración de tenant no existe |
10. Referencias
