144 lines
4.6 KiB
Python
144 lines
4.6 KiB
Python
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
|
|
# a chave da API 'CHAVE_API_DEMO'
|
|
if not os.path.exists("modulos/constantes.py"):
|
|
chave_api = input("Introduza a chave da API Demo de CoinGecko: ")
|
|
if not aux.validarChaveAPI(chave_api):
|
|
print(f"erro na validação da chave API {chave_api}")
|
|
quit()
|
|
else:
|
|
with open("modulos/constantes.py", "w") as f:
|
|
f.write(f"CHAVE_API_DEMO = '{chave_api}'\n")
|
|
|
|
import modulos.coingecko_api as cgapi
|
|
|
|
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",
|
|
"open",
|
|
"preco_maximo",
|
|
"preco_minimo",
|
|
"preco_fecho",
|
|
]
|
|
|
|
portfolio_headers = ["data", "moeda", "quantidade", "movimento"]
|
|
|
|
# introduzir quantidade de criptomoeda detida
|
|
qtd_criptomoeda = float(
|
|
input(f"Introduza valor de {criptomoeda} detida: ").replace(",", ".")
|
|
)
|
|
|
|
# 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
|
|
)
|
|
|
|
# obter volume de negociação
|
|
# exchange_id: str = "coinbase"
|
|
# url, codigo, dados_volume = cgapi.exchange_volume_chart_by_id(exchange_id, "365")
|
|
# print(dados_volume)
|
|
|
|
# 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:
|
|
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(f"dados obtidos com sucesso ({len(lista_dados_coingecko)})")
|
|
else:
|
|
print("erro ao obter dados")
|
|
print(f"erro {codigo}: {dados_ohlc}")
|
|
|
|
# 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(f"{len(dados)} dados carregados")
|
|
|
|
|
|
# preparar dados para plotar
|
|
dados = dados[1:] # remover cabecalho
|
|
for item in dados:
|
|
item[cabecalho[0]] = pd.to_datetime(int(item[cabecalho[0]]), unit="ms")
|
|
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]])
|
|
|
|
# valor mais recente de criptomoeda em eur
|
|
cripto_vs_euro = float(dados[len(dados) - 1][cabecalho[6]])
|
|
# valor em eur da quantidade de criptomoeda detida
|
|
print(
|
|
f"Valor de {qtd_criptomoeda} {criptomoeda} em EUR: \
|
|
{(qtd_criptomoeda * cripto_vs_euro):.2f}€"
|
|
)
|
|
|
|
|
|
# plotar dados com mplfinance
|
|
daily = pd.DataFrame.from_records(dados, index="data_unix")
|
|
|
|
mpf.plot(daily, type="line", style="charles", title=criptomoeda)
|