nada de novo
This commit is contained in:
@ -7,29 +7,29 @@ import mplfinance as mpf
|
|||||||
|
|
||||||
# se não houver ficheiro 'modulos/constantes.py' criar ficheiro com
|
# se não houver ficheiro 'modulos/constantes.py' criar ficheiro com
|
||||||
# a chave da API 'CHAVE_API_DEMO'
|
# a chave da API 'CHAVE_API_DEMO'
|
||||||
if not os.path.exists('modulos/constantes.py'):
|
if not os.path.exists("modulos/constantes.py"):
|
||||||
chave_api = input('Introduza a chave da API Demo de CoinGecko: ')
|
chave_api = input("Introduza a chave da API Demo de CoinGecko: ")
|
||||||
if (not aux.validarChaveAPI(chave_api)):
|
if not aux.validarChaveAPI(chave_api):
|
||||||
print(f'erro na validação da chave API {chave_api}')
|
print(f"erro na validação da chave API {chave_api}")
|
||||||
quit()
|
quit()
|
||||||
else:
|
else:
|
||||||
with open('modulos/constantes.py', 'w') as f:
|
with open("modulos/constantes.py", "w") as f:
|
||||||
f.write(f'CHAVE_API_DEMO = \'{chave_api}\'\n')
|
f.write(f"CHAVE_API_DEMO = '{chave_api}'\n")
|
||||||
|
|
||||||
import modulos.coingecko_api as cgapi
|
import modulos.coingecko_api as cgapi
|
||||||
|
|
||||||
criptomoeda: str = 'bitcoin'
|
criptomoeda: str = "bitcoin"
|
||||||
vs_currency: str = 'eur'
|
vs_currency: str = "eur"
|
||||||
dias: str = '30'
|
dias: str = "30"
|
||||||
|
|
||||||
cabecalho: list = [
|
cabecalho: list = [
|
||||||
'data_unix',
|
"data_unix",
|
||||||
'criptomoeda',
|
"criptomoeda",
|
||||||
'vs_currency',
|
"vs_currency",
|
||||||
'Open',
|
"Open",
|
||||||
'High',
|
"High",
|
||||||
'Low',
|
"Low",
|
||||||
'Close'
|
"Close",
|
||||||
]
|
]
|
||||||
|
|
||||||
historico_precos_header = [
|
historico_precos_header = [
|
||||||
@ -38,45 +38,46 @@ historico_precos_header = [
|
|||||||
"open",
|
"open",
|
||||||
"preco_maximo",
|
"preco_maximo",
|
||||||
"preco_minimo",
|
"preco_minimo",
|
||||||
"preco_fecho"
|
"preco_fecho",
|
||||||
]
|
]
|
||||||
|
|
||||||
portfolio_headers = [
|
portfolio_headers = ["data", "moeda", "quantidade", "movimento"]
|
||||||
"data",
|
|
||||||
"moeda",
|
|
||||||
"quantidade",
|
|
||||||
"movimento"
|
|
||||||
]
|
|
||||||
|
|
||||||
# introduzir quantidade de criptomoeda detida
|
# introduzir quantidade de criptomoeda detida
|
||||||
qtd_criptomoeda = float(
|
qtd_criptomoeda = float(
|
||||||
input(f"Introduza valor de {criptomoeda} detida: ").replace(",", "."))
|
input(f"Introduza valor de {criptomoeda} detida: ").replace(",", "."))
|
||||||
|
|
||||||
# se não houver ficheiro 'dados/{criptomoeda}.csv' criar ficheiro
|
# se não houver ficheiro 'dados/{criptomoeda}.csv' criar ficheiro
|
||||||
caminho_ficheiro_historico_csv = f'./dados/{criptomoeda}.csv'
|
caminho_ficheiro_historico_csv = f"./dados/{criptomoeda}.csv"
|
||||||
if (not os.path.exists(caminho_ficheiro_historico_csv)):
|
if not os.path.exists(caminho_ficheiro_historico_csv):
|
||||||
ficheiro = open(caminho_ficheiro_historico_csv, 'w+', newline='')
|
ficheiro = open(caminho_ficheiro_historico_csv, "w+", newline="")
|
||||||
ficheiroCSV = csv.DictWriter(ficheiro, fieldnames=cabecalho)
|
ficheiroCSV = csv.DictWriter(ficheiro, fieldnames=cabecalho)
|
||||||
ficheiroCSV.writeheader()
|
ficheiroCSV.writeheader()
|
||||||
ficheiro.close()
|
ficheiro.close()
|
||||||
with open(caminho_ficheiro_historico_csv,
|
with open(
|
||||||
'r',
|
caminho_ficheiro_historico_csv, "r", newline=""
|
||||||
newline='') as ficheiro_csv_historico_precos:
|
) as ficheiro_csv_historico_precos:
|
||||||
ficheiroCSV = csv.DictReader(ficheiro_csv_historico_precos)
|
ficheiroCSV = csv.DictReader(ficheiro_csv_historico_precos)
|
||||||
lista_linhas_ficheiro_csv: list = []
|
lista_linhas_ficheiro_csv: list = []
|
||||||
for linha in ficheiroCSV:
|
for linha in ficheiroCSV:
|
||||||
lista_linhas_ficheiro_csv.append(linha)
|
lista_linhas_ficheiro_csv.append(linha)
|
||||||
|
|
||||||
# obter dados de coingecko
|
# obter dados de coingecko
|
||||||
url, codigo, dados_ohlc = cgapi.coin_ohlc_chart_by_id(criptomoeda, vs_currency,
|
url, codigo, dados_ohlc = cgapi.coin_ohlc_chart_by_id(
|
||||||
days=dias, precision=3)
|
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
|
# processar dados coigecko
|
||||||
if codigo == 200:
|
if codigo == 200:
|
||||||
# print(len(dados_ohlc))
|
# print(len(dados_ohlc))
|
||||||
lista_dados_coingecko: list = []
|
lista_dados_coingecko: list = []
|
||||||
for index in range(0, len(dados_ohlc)):
|
for index in range(0, len(dados_ohlc)):
|
||||||
data = time.gmtime(dados_ohlc[index][0]/1000)
|
data = time.gmtime(dados_ohlc[index][0] / 1000)
|
||||||
if data.tm_hour == 0 and data.tm_min == 0:
|
if data.tm_hour == 0 and data.tm_min == 0:
|
||||||
nova_entrada: dict = {}
|
nova_entrada: dict = {}
|
||||||
nova_entrada[cabecalho[0]] = str(dados_ohlc[index][0])
|
nova_entrada[cabecalho[0]] = str(dados_ohlc[index][0])
|
||||||
@ -87,10 +88,10 @@ if codigo == 200:
|
|||||||
nova_entrada[cabecalho[5]] = str(dados_ohlc[index][3])
|
nova_entrada[cabecalho[5]] = str(dados_ohlc[index][3])
|
||||||
nova_entrada[cabecalho[6]] = str(dados_ohlc[index][4])
|
nova_entrada[cabecalho[6]] = str(dados_ohlc[index][4])
|
||||||
lista_dados_coingecko.append(nova_entrada)
|
lista_dados_coingecko.append(nova_entrada)
|
||||||
print(f'dados obtidos com sucesso ({len(lista_dados_coingecko)})')
|
print(f"dados obtidos com sucesso ({len(lista_dados_coingecko)})")
|
||||||
else:
|
else:
|
||||||
print('erro ao obter dados')
|
print("erro ao obter dados")
|
||||||
print(f'erro {codigo}: {dados_ohlc}')
|
print(f"erro {codigo}: {dados_ohlc}")
|
||||||
|
|
||||||
# adicionar items de lista_dados_coingecko se não existirem em
|
# adicionar items de lista_dados_coingecko se não existirem em
|
||||||
# lista_linhas_ficheiro_csv
|
# lista_linhas_ficheiro_csv
|
||||||
@ -98,7 +99,7 @@ else:
|
|||||||
for item_resposta in lista_dados_coingecko:
|
for item_resposta in lista_dados_coingecko:
|
||||||
item_existe: bool = False
|
item_existe: bool = False
|
||||||
for linha_ficheiro_csv in lista_linhas_ficheiro_csv:
|
for linha_ficheiro_csv in lista_linhas_ficheiro_csv:
|
||||||
if (item_resposta == linha_ficheiro_csv):
|
if item_resposta == linha_ficheiro_csv:
|
||||||
item_existe = True
|
item_existe = True
|
||||||
if not item_existe:
|
if not item_existe:
|
||||||
lista_linhas_ficheiro_csv.append(item_resposta)
|
lista_linhas_ficheiro_csv.append(item_resposta)
|
||||||
@ -108,34 +109,35 @@ lista_linhas_ficheiro_csv.sort(key=lambda x: x[cabecalho[0]])
|
|||||||
|
|
||||||
# gravar dados no ficheiro
|
# gravar dados no ficheiro
|
||||||
aux.gravar_dados_ficheiro_csv(
|
aux.gravar_dados_ficheiro_csv(
|
||||||
caminho_ficheiro_historico_csv,
|
caminho_ficheiro_historico_csv, lista_linhas_ficheiro_csv, cabecalho
|
||||||
lista_linhas_ficheiro_csv,
|
)
|
||||||
cabecalho)
|
|
||||||
|
|
||||||
# carregar dados de ficheiro csv e guardar em lista
|
# carregar dados de ficheiro csv e guardar em lista
|
||||||
dados = aux.carregar_dados_ficheiro_csv(
|
dados = aux.carregar_dados_ficheiro_csv(caminho_ficheiro_historico_csv, cabecalho)
|
||||||
caminho_ficheiro_historico_csv,
|
|
||||||
cabecalho)
|
|
||||||
|
|
||||||
print(f'{len(dados)} dados carregados')
|
print(f"{len(dados)} dados carregados")
|
||||||
|
|
||||||
|
|
||||||
# preparar dados para plotar
|
# preparar dados para plotar
|
||||||
dados = dados[1:] # remover cabecalho
|
dados = dados[1:] # remover cabecalho
|
||||||
for item in dados:
|
for item in dados:
|
||||||
item[cabecalho[0]] = pd.to_datetime(int(item[cabecalho[0]]), unit='ms')
|
item[cabecalho[0]] = pd.to_datetime(int(item[cabecalho[0]]), unit="ms")
|
||||||
item[cabecalho[3]] = float(item[cabecalho[3]])
|
item[cabecalho[3]] = float(item[cabecalho[3]])
|
||||||
item[cabecalho[4]] = float(item[cabecalho[4]])
|
item[cabecalho[4]] = float(item[cabecalho[4]])
|
||||||
item[cabecalho[5]] = float(item[cabecalho[5]])
|
item[cabecalho[5]] = float(item[cabecalho[5]])
|
||||||
item[cabecalho[6]] = float(item[cabecalho[6]])
|
item[cabecalho[6]] = float(item[cabecalho[6]])
|
||||||
|
|
||||||
# plotar dados com mplfinance
|
|
||||||
daily = pd.DataFrame.from_records(dados, index='data_unix')
|
|
||||||
|
|
||||||
mpf.plot(daily, type='line', style='charles', title=criptomoeda)
|
|
||||||
|
|
||||||
# valor mais recente de criptomoeda em eur
|
# valor mais recente de criptomoeda em eur
|
||||||
cripto_vs_euro = float(dados[len(dados)-1][cabecalho[6]])
|
cripto_vs_euro = float(dados[len(dados) - 1][cabecalho[6]])
|
||||||
# valor em eur da quantidade de criptomoeda detida
|
# valor em eur da quantidade de criptomoeda detida
|
||||||
print(f"Valor de {qtd_criptomoeda} {criptomoeda} em EUR: {
|
print(
|
||||||
(qtd_criptomoeda * cripto_vs_euro):.2f}€")
|
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)
|
||||||
|
|||||||
@ -5,9 +5,7 @@ import csv
|
|||||||
|
|
||||||
# gravar dados importados para ficheiro csv
|
# gravar dados importados para ficheiro csv
|
||||||
def gravar_dados_ficheiro_csv(
|
def gravar_dados_ficheiro_csv(
|
||||||
nome_ficheiro: str,
|
nome_ficheiro: str, dados_importados: list, campos: list
|
||||||
dados_importados: list,
|
|
||||||
campos: list
|
|
||||||
) -> None:
|
) -> None:
|
||||||
"""
|
"""
|
||||||
Guarda dados em um ficheiro CSV.
|
Guarda dados em um ficheiro CSV.
|
||||||
@ -26,25 +24,20 @@ def gravar_dados_ficheiro_csv(
|
|||||||
"""
|
"""
|
||||||
# validacao de parametros da funcao
|
# validacao de parametros da funcao
|
||||||
if not isinstance(nome_ficheiro, str):
|
if not isinstance(nome_ficheiro, str):
|
||||||
raise ValueError('\'nome_ficheiro\' não é do tipo \'str\'')
|
raise ValueError("'nome_ficheiro' não é do tipo 'str'")
|
||||||
if not isinstance(dados_importados, list):
|
if not isinstance(dados_importados, list):
|
||||||
raise ValueError('\'dados_importados\' não é do tipo \'list\'')
|
raise ValueError("'dados_importados' não é do tipo 'list'")
|
||||||
if not isinstance(campos, list):
|
if not isinstance(campos, list):
|
||||||
raise ValueError('\'campos\' não é do tipo \'list\'')
|
raise ValueError("'campos' não é do tipo 'list'")
|
||||||
|
|
||||||
with open(nome_ficheiro, 'w', newline='') as ficheiro_csv_historico_precos:
|
with open(nome_ficheiro, "w", newline="") as ficheiro_csv_historico_precos:
|
||||||
ficheiroCSV = csv.DictWriter(
|
ficheiroCSV = csv.DictWriter(ficheiro_csv_historico_precos, fieldnames=campos)
|
||||||
ficheiro_csv_historico_precos,
|
|
||||||
fieldnames=campos)
|
|
||||||
ficheiroCSV.writeheader()
|
ficheiroCSV.writeheader()
|
||||||
ficheiroCSV.writerows(dados_importados)
|
ficheiroCSV.writerows(dados_importados)
|
||||||
|
|
||||||
|
|
||||||
# carregar dados de ficheiro csv e guardar em lista
|
# carregar dados de ficheiro csv e guardar em lista
|
||||||
def carregar_dados_ficheiro_csv(
|
def carregar_dados_ficheiro_csv(nome_ficheiro: str, campos: list) -> list:
|
||||||
nome_ficheiro: str,
|
|
||||||
campos: list
|
|
||||||
) -> list:
|
|
||||||
"""
|
"""
|
||||||
Carrega dados de um ficheiro CSV e retorna uma lista de dicionários.
|
Carrega dados de um ficheiro CSV e retorna uma lista de dicionários.
|
||||||
|
|
||||||
@ -62,14 +55,12 @@ def carregar_dados_ficheiro_csv(
|
|||||||
"""
|
"""
|
||||||
# validacao de metodos da funcao
|
# validacao de metodos da funcao
|
||||||
if not isinstance(nome_ficheiro, str):
|
if not isinstance(nome_ficheiro, str):
|
||||||
raise ValueError('\'nome_ficheiro\' não é do tipo \'str\'')
|
raise ValueError("'nome_ficheiro' não é do tipo 'str'")
|
||||||
if not isinstance(campos, list):
|
if not isinstance(campos, list):
|
||||||
raise ValueError('\'campos\' não é do tipo \'list\'')
|
raise ValueError("'campos' não é do tipo 'list'")
|
||||||
|
|
||||||
with open(nome_ficheiro, 'r', newline='') as ficheiro_csv_historico_precos:
|
with open(nome_ficheiro, "r", newline="") as ficheiro_csv_historico_precos:
|
||||||
ficheiroCSV = csv.DictReader(
|
ficheiroCSV = csv.DictReader(ficheiro_csv_historico_precos, fieldnames=campos)
|
||||||
ficheiro_csv_historico_precos,
|
|
||||||
fieldnames=campos)
|
|
||||||
lista_linhas_ficheiro_csv: list = []
|
lista_linhas_ficheiro_csv: list = []
|
||||||
for linha in ficheiroCSV:
|
for linha in ficheiroCSV:
|
||||||
lista_linhas_ficheiro_csv.append(linha)
|
lista_linhas_ficheiro_csv.append(linha)
|
||||||
@ -84,6 +75,6 @@ def validarChaveAPI(chave_api: str) -> bool:
|
|||||||
if len(chave_api) != 27:
|
if len(chave_api) != 27:
|
||||||
return False
|
return False
|
||||||
# chave inválida se primeiros 3 caracteres forem diferentes de 'CG-'
|
# chave inválida se primeiros 3 caracteres forem diferentes de 'CG-'
|
||||||
if len(chave_api) >= 3 and chave_api[0:3] != 'CG-':
|
if len(chave_api) >= 3 and chave_api[0:3] != "CG-":
|
||||||
return False
|
return False
|
||||||
return True
|
return True
|
||||||
|
|||||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user