Examples
This section provides detailed examples of how to use the BCRA API Connector for various tasks.
Fetching Principal Variables
The following example demonstrates how to fetch and visualize the principal variables from the BCRA API.
import logging
import os
from typing import Optional
import matplotlib.pyplot as plt
from bcra_connector import BCRAApiError, BCRAConnector
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
def save_plot(fig, filename: str) -> None:
"""Saves the given matplotlib figure to the docs static images directory."""
static_dir = os.path.abspath(
os.path.join(os.path.dirname(__file__), "..", "docs/build/_static/images")
)
os.makedirs(static_dir, exist_ok=True)
filepath = os.path.join(static_dir, filename)
fig.savefig(filepath)
logger.info(f"Plot saved as '{filepath}'")
def main() -> None:
"""Main function to demonstrate fetching principal variables."""
connector = BCRAConnector(verify_ssl=False)
try:
logger.info("Fetching principal variables/monetary series (v4.0)...")
variables = connector.get_principales_variables()
logger.info(f"Found {len(variables)} variables/series.")
if not variables:
logger.warning("No variables returned from the API.")
return
logger.info("First 5 variables/series:")
for var in variables[:5]:
logger.info(
f"ID: {var.idVariable}, Description: {var.descripcion}, "
f"Category: {var.categoria if var.categoria else 'N/A'}"
)
if var.ultValorInformado is not None and var.ultFechaInformada:
logger.info(
f" Latest value: {var.ultValorInformado} ({var.ultFechaInformada.isoformat()})"
)
else:
logger.info(" Latest value: Not available")
plot_count = min(10, len(variables))
if plot_count > 0:
# Filter variables that have ultValorInformado
plottable_vars = [
v for v in variables[:plot_count] if v.ultValorInformado is not None
]
if plottable_vars:
fig, ax = plt.subplots(figsize=(12, 6))
ax.bar(
[
(v.descripcion[:30] if v.descripcion else "N/A")
+ (f" ({v.categoria[:10]})" if v.categoria else "")
for v in plottable_vars
],
[v.ultValorInformado for v in plottable_vars],
)
ax.set_title(
f"Top {len(plottable_vars)} Principal Variables/Series (v4.0)"
)
ax.set_xlabel("Variables/Series (Category)")
ax.set_ylabel("Value")
plt.xticks(rotation=45, ha="right")
plt.tight_layout()
save_plot(fig, "principal_variables_v4.png")
else:
logger.info("No variables with values to plot.")
else:
logger.info("No variables to plot.")
variable_name_to_search: Optional[str] = "Reservas Internacionales del BCRA"
if not (
variable_name_to_search
and any(
variable_name_to_search.lower() in v.descripcion.lower()
for v in variables
if v.descripcion
)
):
if variables:
first_var_desc = getattr(variables[0], "descripcion", None)
if first_var_desc:
variable_name_to_search = first_var_desc
logger.warning(
f"'Reservas Internacionales del BCRA' not found or initial search term was None, "
f"using first variable: '{variable_name_to_search}' for history example."
)
else:
variable_name_to_search = None
else:
variable_name_to_search = None
if variable_name_to_search:
try:
logger.info(f"Fetching history for: '{variable_name_to_search}'")
history = connector.get_variable_history(
variable_name_to_search, days=30, limit=15
)
logger.info(
f"Historical data for '{variable_name_to_search}' (last 30 days, limit 15):"
)
if history:
for data_point in history[-5:]:
logger.info(
f" {data_point.fecha.isoformat()}: {data_point.valor}"
)
else:
logger.info(
f"No historical data returned for '{variable_name_to_search}'."
)
except ValueError as e:
logger.error(
f"Error fetching variable history for '{variable_name_to_search}': {str(e)}"
)
else:
logger.info(
"Skipping variable history example as no variable name was determined."
)
except BCRAApiError as e:
logger.error(f"API Error occurred: {str(e)}")
except Exception as e:
logger.error(f"Unexpected error occurred: {str(e)}", exc_info=True)
if __name__ == "__main__":
main()
This script will generate a bar plot of the top 10 principal variables:
Retrieving Historical Data
This example shows how to retrieve historical data for a specific variable and plot it.
import logging
import os
from datetime import datetime, timedelta
import matplotlib.pyplot as plt
import numpy as np
from bcra_connector import BCRAApiError, BCRAConnector
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
def save_plot(fig, filename: str) -> None:
"""Saves the given matplotlib figure to the docs static images directory."""
static_dir = os.path.abspath(
os.path.join(os.path.dirname(__file__), "..", "docs/build/_static/images")
)
os.makedirs(static_dir, exist_ok=True)
filepath = os.path.join(static_dir, filename)
fig.savefig(filepath)
logger.info(f"Plot saved as '{filepath}'")
def main() -> None:
"""Main function to demonstrate fetching historical data for a variable."""
connector = BCRAConnector(verify_ssl=False)
try:
variable_name_to_fetch = "Reservas Internacionales del BCRA"
target_variable = connector.get_variable_by_name(variable_name_to_fetch)
if not target_variable:
logger.warning(
f"Variable '{variable_name_to_fetch}' not found by name. Trying first available variable."
)
all_variables = connector.get_principales_variables()
if not all_variables:
logger.error("No variables found at all. Cannot proceed.")
return
target_variable = all_variables[0]
logger.info(
f"Using variable ID: {target_variable.idVariable} ({target_variable.descripcion}) for demonstration."
)
variable_id_to_use = target_variable.idVariable
display_variable_name = target_variable.descripcion
end_date = datetime.now()
start_date = end_date - timedelta(days=90)
limit_param = 50
offset_param = 0
logger.info(
f"Fetching data for '{display_variable_name}' (ID: {variable_id_to_use}) "
f"from {start_date.date().isoformat()} to {end_date.date().isoformat()} with limit={limit_param}, offset={offset_param}..."
)
response_data = connector.get_datos_variable(
variable_id_to_use,
desde=start_date,
hasta=end_date,
limit=limit_param,
offset=offset_param,
)
datos_list = response_data.results
metadata = response_data.metadata
# In v4.0, results is a list of DatosVariable objects, each with a detalle array
# Flatten all detalle arrays into a single list
all_data_points = []
for datos_variable in datos_list:
all_data_points.extend(datos_variable.detalle)
logger.info(
f"Fetched {len(all_data_points)} data points from {len(datos_list)} result groups. "
f"Total available according to metadata: {metadata.resultset.count}. "
f"Offset: {metadata.resultset.offset}, Limit: {metadata.resultset.limit}."
)
if not all_data_points:
logger.warning(
f"No data points returned for '{display_variable_name}'. Cannot plot or show details."
)
return
logger.info("Last 5 data points from the fetched data:")
for dato in all_data_points[-5:]:
logger.info(f" Date: {dato.fecha.isoformat()}, Value: {dato.valor}")
fig, ax = plt.subplots(figsize=(12, 6))
dates = [
datetime.combine(dato.fecha, datetime.min.time())
for dato in all_data_points
]
values = [dato.valor for dato in all_data_points]
ax.plot_date(np.array(dates), np.array(values), "-")
ax.set_title(
f"'{display_variable_name}\\n(Page with limit={limit_param}, offset={offset_param} in last 90 days)"
)
ax.set_xlabel("Date")
ax.set_ylabel("Value")
plt.xticks(rotation=45, ha="right")
plt.tight_layout()
save_plot(fig, f"variable_{variable_id_to_use}_data_v4_paginated.png")
except BCRAApiError as e:
logger.error(f"API Error occurred: {str(e)}")
except ValueError as e:
logger.error(f"Value Error: {str(e)}")
except Exception as e:
logger.error(f"Unexpected error occurred: {str(e)}", exc_info=True)
if __name__ == "__main__":
main()
The script generates a line plot of the variable’s values over time:
Getting Latest Values
Here’s how to fetch and compare the latest values for multiple variables.
import logging
import os
import matplotlib.pyplot as plt
from bcra_connector import BCRAApiError, BCRAConnector
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
def save_plot(fig, filename: str) -> None:
"""Saves the given matplotlib figure to the docs static images directory."""
static_dir = os.path.abspath(
os.path.join(os.path.dirname(__file__), "..", "docs/build/_static/images")
)
os.makedirs(static_dir, exist_ok=True)
filepath = os.path.join(static_dir, filename)
fig.savefig(filepath)
logger.info(f"Plot saved as '{filepath}'")
def main() -> None:
"""Main function to demonstrate fetching latest values."""
connector = BCRAConnector(verify_ssl=False)
variable_names_to_check = [
"Reservas Internacionales del BCRA",
"Base Monetaria",
]
try:
all_variables = connector.get_principales_variables()
if not all_variables:
logger.error("No variables returned from API. Cannot get latest values.")
return
current_variable_names = []
for name_to_check in variable_names_to_check:
if any(
name_to_check.lower() in v.descripcion.lower() for v in all_variables
):
current_variable_names.append(name_to_check)
else:
logger.warning(
f"Predefined variable '{name_to_check}' not found in API results."
)
if not current_variable_names:
logger.info("Using first few available distinct variables for example.")
seen_ids = set()
for v_api in all_variables:
if v_api.idVariable not in seen_ids:
current_variable_names.append(v_api.descripcion)
seen_ids.add(v_api.idVariable)
if len(current_variable_names) >= 3: # Limit to 3 for example
break
if not current_variable_names:
logger.error(
"Could not determine any variables to check for latest values."
)
return
except BCRAApiError as e:
logger.error(f"Could not fetch variable list to select examples: {e}")
return
except Exception as e:
logger.error(f"Unexpected error fetching variable list: {e}", exc_info=True)
return
latest_values_data = []
for name_or_desc in current_variable_names:
try:
logger.info(f"Fetching latest value for '{name_or_desc}'...")
variable_obj = connector.get_variable_by_name(name_or_desc)
if not variable_obj:
logger.warning(
f"Variable '{name_or_desc}' not found by name during latest value fetch."
)
continue
# v4.0: get_latest_value() returns DetalleMonetaria (not DatosVariable)
latest_data_point = connector.get_latest_value(variable_obj.idVariable)
logger.info(
f" ID: {variable_obj.idVariable}, Value: {latest_data_point.valor}, "
f"Date: {latest_data_point.fecha.isoformat()}, Category: {getattr(variable_obj, 'categoria', 'N/A')}"
)
latest_values_data.append(
(variable_obj.descripcion, latest_data_point.valor)
)
except BCRAApiError as e:
logger.error(f"API Error for '{name_or_desc}': {str(e)}")
except ValueError as e:
logger.error(f"Value Error for '{name_or_desc}': {str(e)}")
except Exception as e:
logger.error(
f"Unexpected error for '{name_or_desc}': {str(e)}", exc_info=True
)
if latest_values_data:
fig, ax = plt.subplots(figsize=(12, 7))
plot_names = [item[0][:40] for item in latest_values_data]
plot_values = [item[1] for item in latest_values_data]
ax.bar(plot_names, plot_values)
ax.set_title("Latest Values for Different Variables/Series (v4.0)")
ax.set_xlabel("Variable/Series")
ax.set_ylabel("Value")
plt.xticks(rotation=45, ha="right", fontsize=9)
plt.tight_layout()
save_plot(fig, "latest_values_v4.png")
else:
logger.warning("No data to plot for latest values.")
if __name__ == "__main__":
main()
This example creates a bar plot comparing the latest values:
Error Handling
The following example demonstrates how the connector handles various error scenarios.
import logging
from datetime import datetime, timedelta
from typing import Any, Callable, Type
from bcra_connector import BCRAApiError, BCRAConnector
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
def test_case(
description: str,
func: Callable[[], Any],
expected_exception: Type[BaseException] = Exception,
) -> None:
"""Helper function to run a test case and log results."""
logger.info(f"\n--- Test case: {description} ---")
try:
result = func()
logger.info(f"Test completed. Result (if any): {result}")
except expected_exception as e:
logger.info(f"Expected exception caught: {type(e).__name__}: {str(e)}")
except Exception as e:
logger.error(
f"Unexpected exception raised: {type(e).__name__}: {str(e)} "
f"(Expected {expected_exception.__name__} or success to not raise)",
exc_info=True,
)
def main() -> None:
"""Main function to demonstrate error handling test cases."""
connector = BCRAConnector(verify_ssl=False, debug=True)
test_case(
"Invalid variable ID for get_latest_value",
lambda: connector.get_latest_value(9999999),
expected_exception=BCRAApiError,
)
def invalid_date_order():
return connector.get_datos_variable(
1, datetime(2023, 1, 10), datetime(2023, 1, 1)
)
test_case(
"Invalid date order (desde > hasta) for get_datos_variable",
invalid_date_order,
expected_exception=ValueError,
)
def invalid_limit_low():
return connector.get_datos_variable(1, limit=5)
test_case(
"Invalid limit (too low) for get_datos_variable",
invalid_limit_low,
expected_exception=ValueError,
)
def invalid_limit_high():
return connector.get_datos_variable(1, limit=3001)
test_case(
"Invalid limit (too high) for get_datos_variable",
invalid_limit_high,
expected_exception=ValueError,
)
def future_date_query():
today = datetime.now()
future_start = today + timedelta(days=30)
future_end = today + timedelta(days=60)
response = connector.get_datos_variable(1, future_start, future_end)
return f"Results count: {len(response.results)}"
test_case(
"Query with future date range (API behavior test)",
future_date_query,
expected_exception=BCRAApiError,
)
test_case(
"Non-existent variable name for get_variable_history",
lambda: connector.get_variable_history("This Variable Does Not Exist For Sure"),
expected_exception=ValueError,
)
def simulate_api_error_for_datos():
return connector.get_datos_variable(
9999999, datetime.now() - timedelta(days=1), datetime.now()
)
test_case(
"API error for non-existent ID with get_datos_variable",
simulate_api_error_for_datos,
expected_exception=BCRAApiError,
)
test_case(
"Invalid currency code for get_currency_evolution",
lambda: connector.get_currency_evolution("XZY"),
expected_exception=BCRAApiError,
)
test_case(
"Invalid currency pair for get_currency_pair_evolution",
lambda: connector.get_currency_pair_evolution("XZY", "ABC"),
expected_exception=BCRAApiError,
)
test_case(
"Generate report for non-existent variable",
lambda: connector.generate_variable_report("This Variable Also Does Not Exist"),
expected_exception=ValueError,
)
test_case(
"Correlation between non-existent variables",
lambda: connector.get_variable_correlation(
"NonExistentVarAlpha", "NonExistentVarBeta"
),
expected_exception=ValueError,
)
logger.info("\nError handling example script finished.")
if __name__ == "__main__":
main()
Connector Configuration
This example showcases different configuration options for the BCRA API Connector.
import logging
from datetime import datetime, timedelta
from bcra_connector import BCRAApiError, BCRAConnector
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
def test_connection(connector: BCRAConnector, description: str) -> None:
"""Tests basic connectivity and data fetching with the given connector."""
logger.info(f"\n--- Testing: {description} ---")
try:
variables = connector.get_principales_variables()
logger.info(
f"Successfully fetched {len(variables)} principal variables/series."
)
if not variables:
logger.warning("No principal variables returned.")
return
first_var = variables[0]
logger.info(
f"First variable: ID={first_var.idVariable}, Desc='{first_var.descripcion}', Cat='{first_var.categoria}'"
)
variable_id_to_test = first_var.idVariable
end_date = datetime.now()
start_date = end_date - timedelta(days=7)
logger.info(f"Fetching data for variable ID {variable_id_to_test}...")
response_data = connector.get_datos_variable(
variable_id_to_test, desde=start_date, hasta=end_date, limit=10
)
datos_list = response_data.results
logger.info(
f"Successfully fetched {len(datos_list)} data points for variable ID {variable_id_to_test}."
)
if datos_list:
latest_point = max(datos_list, key=lambda x: x.fecha)
logger.info(
f" Latest fetched data point: Date: {latest_point.fecha.isoformat()}, Value: {latest_point.valor}"
)
else:
logger.info(
f" No data points returned for variable ID {variable_id_to_test} in the specified range/limit."
)
except BCRAApiError as e:
logger.error(f"API Error occurred during '{description}': {str(e)}")
except ValueError as e:
logger.error(f"Value Error occurred during '{description}': {str(e)}")
except Exception as e:
logger.error(
f"Unexpected error during '{description}': {str(e)}", exc_info=True
)
def main() -> None:
"""Main function to demonstrate different connector configurations."""
logger.info("Starting connector configuration examples...")
logger.info(
"\nNOTE: Default connector test (SSL ON) might fail if system certs/proxy are not set up for api.bcra.gob.ar."
)
try:
connector_default = BCRAConnector()
test_connection(connector_default, "Default connector (SSL verification ON)")
except BCRAApiError as e:
logger.error(f"Default connector (SSL ON) API error: {e}")
except Exception as e:
logger.error(f"Unexpected error with default connector: {e}", exc_info=True)
logger.info(
"\nWARNING: The following tests disable SSL verification. This is not recommended for production."
)
connector_no_ssl = BCRAConnector(verify_ssl=False)
test_connection(connector_no_ssl, "Connector with SSL verification disabled")
connector_debug = BCRAConnector(verify_ssl=False, debug=True)
test_connection(
connector_debug, "Connector with SSL verification disabled and debug mode ON"
)
connector_en = BCRAConnector(verify_ssl=False, language="en-US", debug=True)
test_connection(
connector_en, "Connector with English language (en-US) and debug ON"
)
logger.info(
"\nConnector configuration examples finished. "
"In a production environment, always prefer SSL verification (verify_ssl=True)."
)
if __name__ == "__main__":
main()
Cheques Module Usage
This example demonstrates how to interact with the Cheques API, including fetching financial entities and checking for reported checks.
import logging
from bcra_connector import BCRAApiError, BCRAConnector
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
def main():
"""Main function to demonstrate Checks API usage."""
connector = BCRAConnector(verify_ssl=False)
try:
# 1. List Entities
logger.info("Fetching financial entities...")
entities = connector.get_entidades()
logger.info(f"Found {len(entities)} entities.")
if not entities:
logger.warning("No entities found.")
return
logger.info("First 5 entities:")
for entity in entities[:5]:
logger.info(f" Code: {entity.codigo_entidad}, Name: {entity.denominacion}")
# 2. Check a specific check (Simulated)
# We'll use the first entity found
target_entity = entities[0]
check_number = 123456 # Dummy check number
logger.info(
f"Checking check status {check_number} for {target_entity.denominacion} (Code: {target_entity.codigo_entidad})..."
)
# Method A: check_denunciado helper (returns boolean)
try:
is_denounced = connector.check_denunciado(
target_entity.denominacion, check_number
)
logger.info(
f"Check {check_number} denounced status (via check_denunciado): {is_denounced}"
)
except Exception as e:
logger.warning(f"Helper check_denunciado failed: {e}")
# Method B: get_cheque_denunciado (returns Cheque object or raises error)
try:
cheque_info = connector.get_cheque_denunciado(
target_entity.codigo_entidad, check_number
)
logger.info(f"Cheque info found: {cheque_info}")
except BCRAApiError as e:
# It is expected to fail with 404 if the check is not denounced/found
logger.info(
f"get_cheque_denunciado result: {e} (This usually means the check is not denounced)"
)
except Exception as e:
logger.error(f"Unexpected error: {e}", exc_info=True)
if __name__ == "__main__":
main()
Exchange Statistics Usage
This example shows how to use the Exchange Statistics (Estadísticas Cambiarias) API to fetch currencies, quotations, and evolution data.
import logging
import os
import matplotlib.pyplot as plt
from bcra_connector import BCRAApiError, BCRAConnector
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)
def save_plot(fig, filename: str) -> None:
"""Saves the given matplotlib figure to the docs static images directory."""
static_dir = os.path.abspath(
os.path.join(os.path.dirname(__file__), "..", "docs/build/_static/images")
)
os.makedirs(static_dir, exist_ok=True)
filepath = os.path.join(static_dir, filename)
fig.savefig(filepath)
logger.info(f"Plot saved as '{filepath}'")
def main():
"""Main function to demonstrate Exchange Statistics API."""
connector = BCRAConnector(verify_ssl=False)
try:
# 1. Get Currencies
logger.info("Fetching available currencies...")
currencies = connector.get_divisas()
logger.info(f"Found {len(currencies)} currencies.")
logger.info("First 5 currencies:")
for curr in currencies[:5]:
logger.info(f" Code: {curr.codigo}, Name: {curr.denominacion}")
# 2. Get Latest Quotations
logger.info("Fetching latest quotations...")
quotations = connector.get_cotizaciones()
date_str = quotations.fecha.isoformat() if quotations.fecha else "Unknown Date"
logger.info(f"Quotations for date: {date_str}")
logger.info("First 5 quotations:")
for detail in quotations.detalle[:5]:
logger.info(
f" {detail.codigo_moneda} ({detail.descripcion}): {detail.tipo_cotizacion}"
)
# 3. Evolution of USD
target_currency = "USD"
days_to_fetch = 30
logger.info(
f"Fetching evolution of {target_currency} for last {days_to_fetch} days..."
)
try:
usd_evolution = connector.get_currency_evolution(
target_currency, days=days_to_fetch
)
dates = []
values = []
for c in usd_evolution:
if c.fecha:
# Find the specific currency detail in the list
for d in c.detalle:
if d.codigo_moneda == target_currency:
dates.append(c.fecha)
values.append(d.tipo_cotizacion)
break
if dates:
# Sort by date just in case
sorted_pairs = sorted(zip(dates, values))
dates_list, values_list = zip(*sorted_pairs)
fig, ax = plt.subplots(figsize=(10, 6))
ax.plot(list(dates_list), list(values_list), marker="o", linestyle="-")
ax.set_title(f"{target_currency} Evolution (Last {days_to_fetch} Days)")
ax.set_xlabel("Date")
ax.set_ylabel("Rate (ARS)")
plt.xticks(rotation=45, ha="right")
plt.tight_layout()
save_plot(fig, "usd_evolution.png")
else:
logger.warning(f"No data points found for {target_currency} evolution.")
except BCRAApiError as e:
logger.error(f"Failed to fetch evolution: {e}")
except Exception as e:
logger.error(f"Unexpected error: {e}", exc_info=True)
if __name__ == "__main__":
main()
These examples provide a comprehensive overview of the BCRA API Connector’s capabilities and usage patterns.