-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathIOLClient.py
171 lines (129 loc) · 6.78 KB
/
IOLClient.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
from typing import Dict, Any
import pandas as pd
import yfinance as yf
from CommonBroker import CommonBroker
class IOLClient(CommonBroker):
def __init__(self):
super().__init__()
def _obtener_simbolo(self, row) -> str:
"""Procesa el símbolo para tener un formato consistente"""
simbolo = str(row['Simbolo']).split()[0]
print("Simbolo a operar:", simbolo)
print("Moneda:", row['Moneda'])
if str(row['Moneda']).upper() == "US$":
# caso Citigroup "C.D"
if simbolo.endswith(".D"):
return simbolo[:-2]
# Elimina la "D" de los tickers de Cedears en dólares
# Ej "NVDAD" en dolares -> NVDA en dolares
# Si no hacemos esto, veremos el portfolio con tickers repetidos, NVDA y NVDAD por ejemplo
if simbolo.endswith("D"):
return simbolo[:-1]
return simbolo
def _process_transactions(self, df: pd.DataFrame) -> pd.DataFrame:
"""Procesa las transacciones y calcula el portfolio actual"""
# TODO: No estamos considerando Splits!
# El archivo importado de IOL no tiene información sobre splits, por lo que no podemos ajustar las cantidades
df['Cantidad'] = pd.to_numeric(df['Cantidad'], errors='coerce')
df['Precio Ponderado'] = pd.to_numeric(df['Precio Ponderado'], errors='coerce')
# Evitar notación científica
pd.set_option('display.float_format', lambda x: '%.2f' % x)
# Convertir precios a USD si están en ARS
df['Precio USD'] = df.apply(
lambda row: self.pesos_to_usdCCL(row['Precio Ponderado'])
if row['Moneda'].upper() == 'AR$'
else row['Precio Ponderado'],
axis=1
)
df['Monto'] = df['Cantidad'] * df['Precio USD']
# Ajustar cantidades según tipo de transacción
df['Cantidad_Ajustada'] = df.apply(lambda row:
-row['Cantidad'] if row['Tipo Transacción'] in ['Venta', 'Rescate FCI']
else row['Cantidad'], axis=1)
# Procesar símbolos y filtrar cauciones
df['Simbolo'] = df.apply(self._obtener_simbolo, axis=1)
df = df[~df['Simbolo'].str.contains('Caución', na=False, case=False)]
# Crear DataFrame solo con compras para calcular precio promedio
compras_df = df[df['Tipo Transacción'].isin(['Compra', 'Suscripción FCI'])].copy()
# Calcular precio promedio ponderado de compras en USD
precios_promedio = (compras_df.groupby('Simbolo')
.agg({
'Monto': 'sum',
'Cantidad': 'sum'
})
.assign(Precio_Promedio=lambda x: x['Monto'] / x['Cantidad'])
['Precio_Promedio'])
# Calcular posiciones actuales
positions = df.groupby('Simbolo').agg({
'Descripción': 'first',
'Cantidad_Ajustada': 'sum',
'Mercado': 'first'
}).reset_index()
# Agregar precio promedio de compra a las posiciones
positions = positions.merge(
precios_promedio.reset_index(),
on='Simbolo',
how='left'
).rename(columns={
'Cantidad_Ajustada': 'Quantity',
'Precio_Promedio': 'Price (USD)',
'Descripción': 'Name'
})
# Filtrar solo posiciones actuales con cantidad distinta de 0
positions = positions[positions['Quantity'] != 0]
print("\nPosiciones actuales (valores exactos):")
for _, row in positions.iterrows():
print(f"{row['Simbolo']}: {row['Quantity']} @ {row['Price (USD)']}")
return positions
def _calculate_portfolio(self, positions: pd.DataFrame) -> pd.DataFrame:
"""Convierte las posiciones al formato estándar"""
portfolio = pd.DataFrame(columns=self.PORTFOLIO_COLUMNS)
for _, row in positions.iterrows():
portfolio = pd.concat([portfolio, pd.DataFrame([{
'Ticker': row['Simbolo'],
'Name': row['Name'],
'Price (USD)': row['Price (USD)'],
'Quantity': row['Quantity'],
'Price Change (USD)': 0,
'Price Change (%)': 0,
'Total Value (USD)': row['Price (USD)'] * row['Quantity'],
'Market': row['Mercado']
}])], ignore_index=True)
return portfolio
def _obtener_precio_actual(self, ticker: str, market: str) -> tuple:
"""Obtiene el precio actual y previous close de un ticker"""
try:
if market.upper() == 'BCBA':
ticker = f"{ticker}.BA"
stock = yf.Ticker(ticker)
current_price = stock.fast_info.get('lastPrice')
prev_close = stock.fast_info.get('previousClose')
return float(current_price), float(prev_close) # type: ignore
except Exception as e:
print(f"Error obteniendo precio para {ticker}: {e}")
return 0.0, 0.0
def _set_price_changes(self, portfolio: pd.DataFrame) -> pd.DataFrame:
"""Setea los cambios de precio (% y USD) en el portfolio"""
for index, row in portfolio.iterrows():
# Obtener precio actual según el mercado y convertir a USD si es necesario
current_price, prev_close = self._obtener_precio_actual(row['Ticker'], row['Market'])
print(f"Precio actual de {row['Ticker']}: {current_price}")
if current_price > 0:
# Convertir a USD solo si el precio viene del mercado BCBA
if row['Market'].upper() == 'BCBA':
current_price = self.pesos_to_usdCCL(current_price)
prev_close = self.pesos_to_usdCCL(prev_close)
intraday_change_usd = current_price - prev_close
intraday_change_pct = (intraday_change_usd / prev_close) * 100 if prev_close > 0 else 0
portfolio.at[index, 'Price (USD)'] = current_price
portfolio.at[index, 'Price Change (USD)'] = intraday_change_usd
portfolio.at[index, 'Price Change (%)'] = intraday_change_pct
portfolio.at[index, 'Total Value (USD)'] = current_price * row['Quantity']
return portfolio.drop('Market', axis=1)
def get_portfolio(self, file_path: str) -> pd.DataFrame:
"""Lee y procesa un archivo de operaciones de IOL"""
df = self.read_file(file_path)
positions = self._process_transactions(df)
# Convertir a formato estándar
portfolio = self._calculate_portfolio(positions)
return self._set_price_changes(portfolio)