Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 44 additions & 0 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
name: CI

on:
push:
branches: [ main ]
pull_request:
branches: [ main ]

jobs:
build:
runs-on: ubuntu-latest
strategy:
matrix:
python-version: ["3.12"]

steps:
- uses: actions/checkout@v4

- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v5
with:
python-version: ${{ matrix.python-version }}

- name: Install Poetry
uses: snok/install-poetry@v1
with:
virtualenvs-create: true
virtualenvs-in-project: true

- name: Install dependencies
working-directory: ./pyrb3
run: poetry install --no-interaction

- name: Lint with ruff
working-directory: ./pyrb3
run: poetry run ruff check .

- name: Type check with mypy
working-directory: ./pyrb3
run: poetry run mypy .

- name: Test with pytest
working-directory: ./pyrb3
run: poetry run pytest
100 changes: 100 additions & 0 deletions download_index.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,100 @@
import sys
import json
import argparse
import tempfile
import csv
from pathlib import Path
from datetime import date

# Add the pyrb3 source directory to the Python path
sys.path.insert(0, str(Path(__file__).resolve().parent / "pyrb3" / "src"))

from pyrb3.template import Template
from pyrb3.downloaders import stock_indexes_statistics_download

def process_data(data, index_name, year):
"""Processes the raw index data into a time series format."""
processed_rows = []
if not data.get("results"):
return processed_rows

for item in data.get("results", []):
if item.get("day") is None:
continue
day = int(item.get("day"))
for i in range(1, 13):
month_key = f"rateValue{i}"
if month_key in item and item[month_key] is not None:
try:
value_str = item[month_key].replace(",", "")
value = float(value_str)
ref_date = date(year, i, day)
processed_rows.append([ref_date.isoformat(), index_name, value])
except (ValueError, TypeError):
continue
return processed_rows

def main():
"""
Downloads the entire historical data for a given B3 index and saves it to a CSV file.
"""
parser = argparse.ArgumentParser(description="Download B3 index historical data.")
parser.add_argument("index", help="The name of the index (e.g., IBOV, SMLL).")
args = parser.parse_args()

template = Template("b3-indexes-historical-data")
all_rows = []

# Let's assume the current year is 2024 for stable testing
current_year = 2024
consecutive_empty_years = 0

print(f"Starting download for the entire history of index '{args.index}'...")

for year in range(current_year, 1999, -1):
with tempfile.NamedTemporaryFile(delete=False, mode='w+', suffix=".json") as temp_file:
temp_path = Path(temp_file.name)

print(f"Downloading data for {year}...")
success = stock_indexes_statistics_download(
template, temp_path, index=args.index, year=str(year)
)

if success:
with open(temp_path, "r", encoding="utf8") as f:
raw_data = json.load(f)

processed_rows = process_data(raw_data, args.index, year)

if processed_rows:
all_rows.extend(processed_rows)
consecutive_empty_years = 0
else:
consecutive_empty_years += 1

temp_path.unlink()
else:
consecutive_empty_years += 1
temp_path.unlink()

# Stop if we don't find data for 3 consecutive years
if consecutive_empty_years >= 3:
print(f"No data found for {consecutive_empty_years} consecutive years. Stopping.")
break

if all_rows:
# Sort data by date
all_rows.sort(key=lambda x: x[0])

output_filename = f"{args.index}_history.csv"
with open(output_filename, "w", newline="", encoding="utf8") as csv_file:
writer = csv.writer(csv_file)
writer.writerow(["refdate", "index", "value"])
writer.writerows(all_rows)
print(f"Historical data successfully saved to '{output_filename}'")
else:
print(f"No historical data found for index '{args.index}'.")


if __name__ == "__main__":
main()
44 changes: 44 additions & 0 deletions list_indexes.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
import sys
import json
from pathlib import Path
import tempfile

# Add the pyrb3 source directory to the Python path
sys.path.insert(0, str(Path(__file__).resolve().parent / "pyrb3" / "src"))

from pyrb3.template import Template
from pyrb3.downloaders import stock_indexes_composition_download

def main():
"""
Retrieves and prints the list of available B3 indexes.
"""
template = Template("b3-indexes-composition")

with tempfile.NamedTemporaryFile(delete=False) as temp_file:
temp_path = Path(temp_file.name)

if stock_indexes_composition_download(template, temp_path):
with open(temp_path, "r", encoding="latin1") as f:
data = json.load(f)

all_indexes = set()
for item in data.get("results", []):
indexes = item.get("indexes", "").split(",")
for index in indexes:
if index:
all_indexes.add(index.strip())

if all_indexes:
print("Available indexes:")
for index in sorted(list(all_indexes)):
print(f"- {index}")
else:
print("No indexes found.")

temp_path.unlink() # Clean up the temporary file
else:
print("Failed to download index composition data.")

if __name__ == "__main__":
main()
Empty file added pyrb3/README.md
Empty file.
Loading
Loading