criação de ficheiro 'aux.py' com funções auxiliares (como gravar e carregar ficheiros);

aidionado gráfico de velas

crypto_portfolio.py

@@ -1,7 +1,12 @@
 import time
 import os
+import csv
+import modulos.aux as aux
+import pandas as pd
+import mplfinance as mpf
 
-# se não houver ficheiro 'modulos/constantes.py' criar ficheiro com 'CHAVE_API_DEMO'
+# se não houver ficheiro 'modulos/constantes.py' criar ficheiro com 
+# a chave da API 'CHAVE_API_DEMO'
 if not os.path.exists('modulos/constantes.py'):
     with open('modulos/constantes.py', 'w') as f:
         chave_api = input('Introduza a chave da API Demo de CoinGecko: ')
@@ -13,10 +18,20 @@ criptomoeda: str = 'bitcoin'
 vs_currency: str = 'eur'
 dias: str = '30'
 
+cabecalho: list = [
+    'data_unix',
+    'criptomoeda',
+    'vs_currency',
+    'Open',
+    'High',
+    'Low',
+    'Close'
+]
+
 historico_precos_header = [ 
     "data",
     "moeda",
-    "preco_abertura",
+    "open",
     "preco_maximo",
     "preco_minimo",
     "preco_fecho"
@@ -29,15 +44,82 @@ portfolio_headers = [
     "movimento"
 ]
 
-url, codigo, dados_ohlc = cgapi.coin_ohlc_chart_by_id(criptomoeda, vs_currency, dias, precision=3)
+# se não houver ficheiro 'dados/{criptomoeda}.csv' criar ficheiro
+caminho_ficheiro_historico_csv = f'./dados/{criptomoeda}.csv'
+if (not os.path.exists(caminho_ficheiro_historico_csv)):
+    ficheiro = open(caminho_ficheiro_historico_csv, 'w+', newline='')
+    ficheiroCSV = csv.DictWriter(ficheiro, fieldnames=cabecalho)
+    ficheiroCSV.writeheader()
+    ficheiro.close()
+with open(caminho_ficheiro_historico_csv,
+          'r',
+          newline='') as ficheiro_csv_historico_precos:
+    ficheiroCSV = csv.DictReader(ficheiro_csv_historico_precos)
+    lista_linhas_ficheiro_csv: list = []
+    for linha in ficheiroCSV:
+        lista_linhas_ficheiro_csv.append(linha)
 
+# obter dados de coingecko
+url, codigo, dados_ohlc = cgapi.coin_ohlc_chart_by_id(criptomoeda, vs_currency, 
+                                                      days=dias, precision=3)
+
+# processar dados coigecko
 if codigo == 200:
     #print(len(dados_ohlc))
+    lista_dados_coingecko: list = []
     for index in range(0, len(dados_ohlc)):
         data = time.gmtime(dados_ohlc[index][0]/1000)
         if data.tm_hour == 0 and data.tm_min == 0:
-            date_as_string = time.strftime("%d/%m/%Y", data)
-            price = dados_ohlc[index][4]
-            print(f'{date_as_string} -> {price}')
+            nova_entrada: dict = {}
+            nova_entrada[cabecalho[0]] = str(dados_ohlc[index][0])
+            nova_entrada[cabecalho[1]] = criptomoeda
+            nova_entrada[cabecalho[2]] = vs_currency
+            nova_entrada[cabecalho[3]] = str(dados_ohlc[index][1])
+            nova_entrada[cabecalho[4]] = str(dados_ohlc[index][2])
+            nova_entrada[cabecalho[5]] = str(dados_ohlc[index][3])
+            nova_entrada[cabecalho[6]] = str(dados_ohlc[index][4])
+            lista_dados_coingecko.append(nova_entrada)
+    print('data retrieved successfully')
 else:
     print('failed to retrive data')
+
+# adicionar items de lista_dados_coingecko se não existirem em 
+# lista_linhas_ficheiro_csv 
+# (aka: adicionar apenas items novos a lista_linhas_ficheiro_csv)
+for item_resposta in lista_dados_coingecko:
+    item_existe: bool = False
+    for linha_ficheiro_csv in lista_linhas_ficheiro_csv:
+        if (item_resposta == linha_ficheiro_csv):
+            item_existe = True
+    if not item_existe:
+        lista_linhas_ficheiro_csv.append(item_resposta)
+
+# ordenar lista por data
+lista_linhas_ficheiro_csv.sort(key=lambda x: x[cabecalho[0]])
+
+# gravar dados no ficheiro
+aux.gravar_dados_ficheiro_csv(
+    caminho_ficheiro_historico_csv,
+    lista_linhas_ficheiro_csv,
+    cabecalho)
+
+# carregar dados de ficheiro csv e guardar em lista
+dados = aux.carregar_dados_ficheiro_csv(
+    caminho_ficheiro_historico_csv,
+    cabecalho)
+
+#print(dados)
+
+# preparar dados para plotar
+dados = dados[1:] # remover cabecalho
+for item in dados:
+    item[cabecalho[0]] = pd.to_datetime(int(item[cabecalho[0]])/1000)
+    item[cabecalho[3]] = float(item[cabecalho[3]])
+    item[cabecalho[4]] = float(item[cabecalho[4]])
+    item[cabecalho[5]] = float(item[cabecalho[5]])
+    item[cabecalho[6]] = float(item[cabecalho[6]])
+
+# plotar dados com mplfinance
+daily = pd.DataFrame.from_records(dados, index='data_unix')
+
+mpf.plot(daily, type='candle', style='charles', title=criptomoeda)

dados/bitcoin.csv

@@ -1,4 +1,4 @@
-data_unix,criptomoeda,vs_currency,preco_abertura,preco_maximo,preco_minimo,preco_fecho
+data_unix,criptomoeda,vs_currency,Open,High,Low,Close
 1713830400000,bitcoin,eur,62492.696,63079.898,62338.092,62738.591
 1713844800000,bitcoin,eur,62731.622,63007.936,62433.917,62433.917
 1713859200000,bitcoin,eur,62351.893,62669.885,61913.266,62025.768
@@ -161,6 +161,7 @@ data_unix,criptomoeda,vs_currency,preco_abertura,preco_maximo,preco_minimo,preco
 1716120000000,bitcoin,eur,61808.823,62146.377,61512.551,61829.506
 1716134400000,bitcoin,eur,61818.297,61818.297,61273.898,61479.361
 1716148800000,bitcoin,eur,61472.072,61594.942,60698.03,60698.03
+1716163200000,bitcoin,eur,60788.937,61088.408,60761.301,60924.868
 1716177600000,bitcoin,eur,60943.195,61418.882,60803.093,61310.779
 1716192000000,bitcoin,eur,61265.22,61777.561,60869.73,61498.328
 1716206400000,bitcoin,eur,61508.732,61915.705,61378.922,61679.177
@@ -181,7 +182,6 @@ data_unix,criptomoeda,vs_currency,preco_abertura,preco_maximo,preco_minimo,preco
 1716422400000,bitcoin,eur,64477.473,64503.784,63763.614,63912.084
 1716436800000,bitcoin,eur,63859.939,64250.165,63859.939,64010.364
 1716451200000,bitcoin,eur,64082.909,64493.094,64024.978,64328.023
-1716163200000,bitcoin,eur,60788.937,61088.408,60761.301,60924.868
 1716465600000,bitcoin,eur,64373.908,64552.559,64101.853,64552.559
 1716480000000,bitcoin,eur,64483.015,64510.68,62482.214,62810.554
 1716494400000,bitcoin,eur,62849.4,62919.692,61920.421,62253.333
@@ -508,3 +508,9 @@ data_unix,criptomoeda,vs_currency,preco_abertura,preco_maximo,preco_minimo,preco
 1721116800000,bitcoin,eur,59463.668,59712.852,57654.563,57654.563
 1721131200000,bitcoin,eur,57729.996,58593.039,57442.529,58441.367
 1721145600000,bitcoin,eur,58517.942,59731.204,58054.525,59731.204
+1721174400000,bitcoin,eur,59799.733,59906.4,58920.16,59763.545
+1721260800000,bitcoin,eur,59060.704,59105.646,58486.724,58625.722
+1721347200000,bitcoin,eur,58286.158,58829.995,58183.065,58665.955
+1721433600000,bitcoin,eur,61837.039,61864.113,61205.441,61224.607
+1721520000000,bitcoin,eur,61766.449,61929.716,61476.959,61705.047
+1721606400000,bitcoin,eur,61966.574,62736.142,61800.969,62467.873

modulos/aux.py

@@ -0,0 +1,74 @@
+# script com funções auxiliares
+
+import csv
+
+# gravar dados importados para ficheiro csv
+def gravar_dados_ficheiro_csv(
+        nome_ficheiro: str,
+        dados_importados: list,
+        campos: list
+) -> None:
+    """
+    Guarda dados em um ficheiro CSV.
+
+    Args:
+        nome_ficheiro (str): O nome do ficheiro CSV.
+        dados_importados (list): Uma lista de dicionários com os dados a serem 
+            gravados.
+        campos (list): Uma lista com os nomes dos campos a serem gravados.
+
+    Raises:
+        ValueError: Se algum dos argumentos não for do tipo esperado.
+
+    Returns:
+        None
+    """
+    # validacao de parametros da funcao
+    if not isinstance(nome_ficheiro, str):
+        raise ValueError('\'nome_ficheiro\' não é do tipo \'str\'')
+    if not isinstance(dados_importados, list):
+        raise ValueError('\'dados_importados\' não é do tipo \'list\'')
+    if not isinstance(campos, list):
+        raise ValueError('\'campos\' não é do tipo \'list\'')
+    
+    with open(nome_ficheiro, 'w', newline='') as ficheiro_csv_historico_precos:
+        ficheiroCSV = csv.DictWriter(
+            ficheiro_csv_historico_precos,
+            fieldnames=campos)
+        ficheiroCSV.writeheader()
+        ficheiroCSV.writerows(dados_importados)
+
+# carregar dados de ficheiro csv e guardar em lista
+def carregar_dados_ficheiro_csv(
+        nome_ficheiro: str,
+        campos: list
+) -> list:
+    """
+    Carrega dados de um ficheiro CSV e retorna uma lista de dicionários.
+
+    Args:
+        nome_ficheiro (str): O nome do ficheiro CSV.
+        campos (list): Uma lista de nomes de campos correspondentes aos 
+            chaves dos dicionários.
+
+    Raises:
+        ValueError: Se algum dos argumentos for inválido.
+
+    Returns:
+        list: Uma lista de dicionários contendo os dados do ficheiro CSV.
+
+    """
+    # validacao de metodos da funcao
+    if not isinstance(nome_ficheiro, str):
+        raise ValueError('\'nome_ficheiro\' não é do tipo \'str\'')
+    if not isinstance(campos, list):
+        raise ValueError('\'campos\' não é do tipo \'list\'')
+    
+    with open(nome_ficheiro, 'r', newline='') as ficheiro_csv_historico_precos:
+        ficheiroCSV = csv.DictReader(
+            ficheiro_csv_historico_precos,
+            fieldnames=campos)
+        lista_linhas_ficheiro_csv: list = []
+        for linha in ficheiroCSV:
+            lista_linhas_ficheiro_csv.append(linha)
+    return lista_linhas_ficheiro_csv

portfolio.py

@@ -9,6 +9,9 @@
 #   . documentação API: https://docs.coingecko.com/v3.0.1/reference/introduction
 #   . obter OHLC: https://docs.coingecko.com/v3.0.1/reference/coins-id-ohlc
 import requests
+import pandas as pd
+import mplfinance as mpf
+import modulos.aux as aux
 
 url_raiz_API = 'https://api.coingecko.com/api/v3/'
 
@@ -50,7 +53,9 @@ if (not os.path.exists(caminho_ficheiro_historico_csv)):
     ficheiroCSV = csv.DictWriter(ficheiro, fieldnames=cabecalho)
     ficheiroCSV.writeheader()
     ficheiro.close()
-with open(caminho_ficheiro_historico_csv, 'r', newline='') as ficheiro_csv_historico_precos:
+with open(caminho_ficheiro_historico_csv,
+          'r',
+          newline='') as ficheiro_csv_historico_precos:
     ficheiroCSV = csv.DictReader(ficheiro_csv_historico_precos)
     lista_linhas_ficheiro_csv: list = []
     for linha in ficheiroCSV:
@@ -80,23 +85,26 @@ for item_resposta in lista_linhas_resposta:
     if not item_existe:
         lista_linhas_ficheiro_csv.append(item_resposta)
 
-# gravar dados importados
-def gravar_dados_ficheiro_csv(
-        nome_ficheiro: str,
-        dados_importados: list,
-        campos: list = cabecalho
-) -> None:
-    # validacao de parametros da funcao
-    if not isinstance(nome_ficheiro, str):
-        raise ValueError('\'nome_ficheiro\' não é do tipo \'str\'')
-    if not isinstance(dados_importados, list):
-        raise ValueError('\'dados_importados\' não é do tipo \'list\'')
-    if not isinstance(campos, list):
-        raise ValueError('\'campos\' não é do tipo \'list\'')
-    
-    with open(nome_ficheiro, 'w', newline='') as ficheiro_csv_historico_precos:
-        ficheiroCSV = csv.DictWriter(ficheiro_csv_historico_precos, fieldnames=campos)
-        ficheiroCSV.writeheader()
-        ficheiroCSV.writerows(dados_importados)
+aux.gravar_dados_ficheiro_csv(
+    caminho_ficheiro_historico_csv,
+    lista_linhas_ficheiro_csv,
+    cabecalho)
 
-gravar_dados_ficheiro_csv(caminho_ficheiro_historico_csv, lista_linhas_ficheiro_csv)
+dados = aux.carregar_dados_ficheiro_csv(
+    caminho_ficheiro_historico_csv,
+    cabecalho)
+
+print(dados)
+# # mostrar dados com mplfinance
+
+# daily = pd.read_csv(caminho_ficheiro_historico_csv, index_col=0)
+# print(dail)
+
+# daily.index.name = 'data_unix'
+# daily.shape
+# daily.head(3)
+# daily.tail(3)
+
+# # plotar grafico
+
+# mpf.plot(daily)