Análise de Dados com Strava
Como melhorar o desempenho das corridas a partir dos dados e aprendizado de maquina.
Sempre gostei muito do mundo das corridas e desde cedo participei de competições, almejando aumentar meu ritmo e distância, mas diversos fatores impactam na corrida, desde fatores psicológicos até fatores externos. Todavia, com o uso de gadgets, vestíveis e aplicativos — como o Strava — tem sido possível capturar métricas e realizar análises que contribuem para o ganho de performance.
Sumário
- Acessando os dados;
- Data Cleaning;
- Insights;
- Seleção de Caraterísticas;
- Aprendizado de Máquina;
- Conclusão;
- Referências.
Acessando os dados
Como fonte dos dados iremos utilizar o Strava, que é um aplicativo Fremium e que possui uma API, possibilitando que os dados sejam capturados sem muito esforço.
Os passos aqui citados para a inserção na API podem também ser acessados a partir da documentação oficial, mas caso queira ser objetivo os passos principais serão descritos nesse artigo.
O primeiro passo para ter acesso aos dados é criar uma conta no Strava, realizar login e depois criar um aplicativo, o que pode ser realizado através do deste link.
Após o aplicativo ter sido criado, iremos utilizar principalmente o Client ID (destacado em verde) e Client Secret (destacado em azul). Que serão colocados em um arquivo .env do repositório ou pasta em que o projeto será executado.
Depois, será necessário acessar o seguinte link, alterando o campo [CLIENT_ID] pelo valor disponibilizado no seu aplicativo, o qual é exemplificado em destaque verde na imagem acima.
Ao acessar esse link, selecione os dados com que deseja que o aplicativo tenha acesso (como na imagem abaixo) e clique em autorizar.
Essa ação irá alterar gerar uma URI diferente, sendo necessário copiar o código do parâmetro code e colar no .env supracitado. Segue abaixo um URI após a autorização, assim como o arquivo .env deve ficar.
client_id=36431
client_secret=fasfjsfajfjie8293u2jf2j92jf232ije0jje92j
strava_code=2e9fuhef9f2jd293jqd0g8erqt84r802jqd0137qApós a realização desses passos, será necessário executar o seguinte script Python no mesmo ambiente do arquivo .env. Esse script irá atualizar os tokens necessários para a captura dos dados e irá gerar um arquivo JSON com os dados retornados da requisição. Nomeei esse script de create_token.py .
import os
import json
import requests
from dotenv import load_dotenv
load_dotenv()
response = requests.post(
url='https://www.strava.com/oauth/token',
data={
'client_id': int(os.environ.get('client_id')),
'client_secret': os.environ.get('client_secret'),
'code': os.environ.get('strava_code'),
'grant_type': 'authorization_code'
}
)
strava_tokens = response.json()
with open('strava_tokens.json', 'w', encoding='utf-8') as outfile:
json.dump(strava_tokens, outfile)
print(strava_tokens)Com os tokens atualizados, bem como as variáveis de ambiente, basta executar o código para a captura dos dados a partir da API, o qual se encontra abaixo. Aqui é possível perceber que a captura ocorre por paginas, sendo assim foi realizado um loop até que a resposta venha vazia ou ocorra um erro. Vale dizer que antes de executar o script criei duas pastas com os nomes de data e result no mesmo workspace para armazenar os CSVs capturados e o script abaixo foi nomeado de get_activities.py .
import os
import json
import glob
import time
import requests
import pandas as pd
from dotenv import load_dotenv
load_dotenv()
def main():
url = "https://www.strava.com/api/v3/activities"
access_token = get_credentials()
page = 1
print('Getting data from Strava')
while True:
response = get_data(url, access_token, 200, page)
if 'message' in response.columns:
raise Exception('Authorization Error')
if response.empty:
break
save_csv(response, f'data/strava_activities_page_{page}.csv')
page += 1
merge_files('data/', 'result/strava_all_activities.csv')
print('Done Successfully')
def get_credentials():
with open('strava_tokens.json', encoding='utf-8') as json_file:
strava_tokens = json.load(json_file)
if 'expires_at' not in strava_tokens.keys() or strava_tokens['expires_at'] < time.time():
strava_tokens = refresh_credentials(strava_tokens)
return strava_tokens['access_token']
def refresh_credentials(strava_tokens):
response = requests.post(
url='https://www.strava.com/oauth/token',
data={
'client_id': int(os.environ.get('client_id')),
'client_secret': os.environ.get('client_secret'),
'grant_type': 'refresh_token',
'refresh_token': strava_tokens['refresh_token']
}
)
strava_tokens = response.json()
with open('strava_tokens.json', 'w', encoding='utf-8') as outfile:
json.dump(strava_tokens, outfile)
with open('strava_tokens.json', encoding='utf-8') as check:
data = json.load(check)
return data
def get_data(url, access_token, numb_item_page, page):
print(f'Getting data from page {page}')
response = requests.get(
f'{url}?access_token={access_token}&per_page={numb_item_page}&page={page}'
)
response = response.json()
dataframe = pd.json_normalize(response)
return dataframe
def save_csv(dataframe, filename):
print(f'Saving {filename}')
dataframe.to_csv(filename)
def merge_files(path, filename):
print('Merging files')
csv_files = [pd.read_csv(_file)
for _file in glob.glob(os.path.join(path, "*.csv"))]
final_df = csv_files.pop(len(csv_files)-1)
final_df = final_df.append(csv_files)
save_csv(final_df, filename)
if __name__ == '__main__':
main()Com os dados em mãos, VAMOS COMEÇAR AS ANÁLISES e para realizar esse feito utilizaremos um Jupyter Notebook.
Data Cleaning
O primeiro passo é importar as bibliotecas que serão utilizadas para realizar a análise de dados e podem ser classificadas em bibliotecas em geral que serão utilizadas para a sustentação do script, bibliotecas para análise e plotagem dos dataframes e, por fim, bibliotecas para a realização de técnicas de aprendizado de maquina.
# general
import subprocess
import calendar
from geopy.geocoders import Nominatim
# df and plotting
import pandas as pd
import numpy as np
import seaborn as sns
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
# machine learning
from sklearn import preprocessing
from sklearn import metrics
from sklearn.cluster import KMeans
from sklearn.feature_selection import chi2
from sklearn.feature_selection import SelectKBest
from sklearn.feature_selection import RFE
from sklearn.linear_model import LinearRegression
from sklearn.linear_model import SGDClassifier
from sklearn.metrics import accuracy_score
from sklearn.metrics import confusion_matrix, ConfusionMatrixDisplay
from sklearn.model_selection import train_test_split
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler
df = pd.read_csv('result/strava_all_activities.csv')
print('Dataframe Shape:', df.shape)
df.head()
O primeiro contato após realizarmos a leitura do CSV resultante nos revela que o nosso dataframe tem por volta de 58 colunas e com uma quantidade de linhas que pode variar de acordo com os registros de cada atleta. Mas quando olhamos para a qualidade dos dados, percebemos que muitos estão nulos ou não contribuem para a analise, então selecionaremos apenas as colunas que são interessantes para a pesquisa.
null_df = [[col, df[col].isnull().sum()] for col in df.columns]
print('Null Data:', df.isnull().sum().sum())
list(filter(lambda x: x[1]>0, null_df))
selected_columns = ['distance', 'moving_time', 'elapsed_time',
'total_elevation_gain', 'type','sport_type', 'id', 'start_date',
'start_date_local','location_country', 'achievement_count', 'kudos_count',
'comment_count','athlete_count', 'start_latlng',
'end_latlng', 'average_speed', 'max_speed', 'average_cadence',
'average_heartrate', 'max_heartrate', 'elev_high','elev_low',
'upload_id', 'external_id', 'pr_count', 'map.summary_polyline']
df = df[selected_columns]Em um primeiro momento decidi extrair o máximo do que o dataset oferece e busquei remover o mínimo possível de dados, então gerei alguns dados a partir da data, completei alguns campos nulos com valores zerados, label de desconhecido ou com a média e, por fim, gerei alguns dados, padronizando a distância em quilômetros e o tempo em minutos.
df['start_date_local'] = pd.to_datetime(df['start_date_local'], errors='coerce')
df = df.sort_values(by='start_date_local')
df['weekday'] = df['start_date_local'].map(lambda x: x.weekday)
df['start_time'] = df['start_date_local'].dt.time
df['start_time'] = df['start_time'].astype(str)
df['start_date'] = df['start_date_local'].dt.date
df = df.drop('start_date_local', 1)
df.head()
df = df.drop(df[(df.distance < 1) & (df.type == 'Run')].index)
df = df.drop(df[(df.distance < 1) & (df.type == 'Ride')].index)
df = df.drop(df[df.average_speed > 30].index)
df = df.reset_index(drop=True)
df['elev_high'] = df['elev_high'].fillna(value=0)
df['elev_low'] = df['elev_low'].fillna(value=0)
df['upload_id'] = df['upload_id'].fillna(value='unknown')
df['external_id'] = df['external_id'].fillna(value='unknown')
df['map.summary_polyline'] = df['map.summary_polyline'].fillna(value='unknown')
df['average_cadence'] = df['average_cadence'].fillna(value=df['average_cadence'].mean())
df['average_heartrate'] = df['average_heartrate'].fillna(value=df['average_heartrate'].mean())
df['max_heartrate'] = df['max_heartrate'].fillna(value=df['max_heartrate'].mean())
df['moving_time_minutes'] = round(df['moving_time']/60, 2)
df['distance_km'] = round(df['distance'] / 1000, 2)
df['pace'] = df['moving_time_minutes'] / df['distance_km']
df['avg_speed_kmh'] = round(60/df['pace'], 2)
df['max_speed_kmh'] = round(df['max_speed']*3.6, 2)
df['elev'] = df['elev_high'] - df['elev_low']
df['year']= df['start_date'].map(lambda x: x.year)Um caso a se destacar é que eu quis completar os campos de cidade e estado, já que dentre as colunas encontravam-se também campos de Latitude e Longitude inicial do treinamento. Então nesse caso foi utilizada a biblioteca geopy.
def get_city_state_from_value(value):
value = value.replace('[','').replace(']','').split(',')
if value != ['']:
location = geolocator.reverse(', '.join(value))
result = f'{location[0].split(",")[1]}, {location[0].split(",")[4]}'
else:
result = 'unknown'
return resultgeolocator = Nominatim(user_agent="strava_exploration_data")
df['location'] = df['start_latlng'].map(get_city_state_from_value)Por fim, foi realizado um filtro baseado no pace abaixo de 5 min/km, ou seja, criou-se uma coluna que demonstrava todas as corridas que tiveram uma média de 12 km/h podendo utilizar isso como variável resposta em análises futuras.
df['pace_sub_5'] = np.where(df['pace']<=5, True, False)
df.head()
E ao executarmos o df.info() e df.describe().transpose(), foi possível obter os seguintes dados:
Insights
Agora com os dados limpos e adequados podemos executar uma série de análises tentando responder perguntas como: Qual foi a quantidade de registros realizados em cada ano? E segundo o seguinte bloco é possível responder essa questão.
fig = sns.catplot(x='year', hue='type', data=df, kind='count')
fig.fig.suptitle('Exercices by Years')
fig.set_xlabels('Year')
fig.set_ylabels('Effortments')
fig
Lembrando que o Strava realiza registra diversos tipos de atividade, mas nesse caso vamos enfatizar apenas a corrida, sendo assim dentre as variáveis que são influenciadoras na performance das corridas (intrinsecamente), destaco elevação, batimentos e fatores relacionados à velocidade, então quis relaciona-los a outros campos.
A primeira analise foi feita relacionando o tempo de atividade com a elevação do trajeto, em que pode-se concluir que a elevação tende a subir conforme o tempo de atividade ou até mesmo conforme a distância.
runs = df.loc[df['type'] == 'Run']
sns.regplot(x='moving_time_minutes', y = 'elev', data=runs).set_title("Exercice Time vs Elevation")
sns.regplot(x='distance', y = 'elev', data=runs).set_title("Distance vs Elevation")
Mas isso não deve ser levado como regra, visto que em vários casos apresentam-se baixas elevações independente do tempo de atividade ou distância, algo muito comum quando o treino ocorre em pistas de corridas de circuito.
Outra observação em relação ao tempo de movimentação e distância, vale dizer que o dia de treinos mais demorados e longos costumam acontecer no domingo, visto que a disponibilidade para a realização dos treinos é maior.
runs.groupby('weekday').mean()['moving_time_minutes'].plot.bar()
runs.groupby('weekday').mean()['distance'].plot.bar()
Em relação à velocidade média (km/h) foi possível observar que quanto maior o tempo de atividade, menor tende a ser a velocidade média, visto que fatores como cansaço começam a interferir na performance, mas quando observamos a velocidade média em relação à distancia, é possível dizer que essa tende a crescer, mesmo que em pequena escala.
sns.regplot(x='moving_time_minutes', y = 'avg_speed_kmh', data=runs).set_title("Average Speed vs Moving Time")
sns.regplot(x='distance', y = 'avg_speed_kmh', data=runs).set_title("Average Speed vs Distance")
Por fim, no meu caso vale dizer que a velocidade média tem aumentado com o passar do tempo o que é fruto dos treinos e todo esforço dedicado no esporte, como demonstra a imagem abaixo.
fig = plt.figure()
ax1 = fig.add_subplot(111)
x = np.asarray(runs.start_date)
y = np.asarray(runs.average_speed)
ax1.plot_date(x, y)
ax1.set_title('Average Speed over Time')
x2 = mdates.date2num(x)
z=np.polyfit(x2,y,1)
p=np.poly1d(z)
plt.plot(x,p(x2),'r--')
fig.autofmt_xdate(rotation=45)
fig.tight_layout()
fig.show()
Seleção de Características
Quanto a seleção de características, vale dizer que esse é um passo importante para a aplicação de modelos de aprendizado de máquina posteriormente. Métodos estatísticos podem ser utilizados para a seleção, a qual irá revelar quais são as features (ou características) mais importantes para a boa performance do modelo. Assim também poderemos confirmar se as características citadas empiricamente são realmente boas.
A primeira forma para entender mais sobre as características e como elas interagem com a variável resposta é por meio da matriz de correlação, o que pode ser facilmente acessado no pandas por meio do seguinte comando.
corr = runs.corr()
plt.figure(figsize = (12,8))
sns.heatmap(corr, fmt=".2f");
plt.title('Correlation between dataset variables')
plt.show()
A partir dessa matriz, pode-se definir que os valores mais próximos de +1 possuem uma correlação positiva e os valores mais próximos de -1 possuem uma correlação negativa.
Para continuar com as análises, o dataframe de corrida será embaralhado de modo que problemas relacionados a bias e ao aprendizado da sequencia do dataframe sejam prevenidos. Além disso, as características categóricas serão removidas da análise, bem como o id das corridas, visto que esses campos não são validos para o treinamento do modelo e campos que não contribuirão para uma previsão, pois são formados depois da corrida ser finalizada.
runs = runs.sample(frac=1).reset_index(drop=True)
categorical_cols = [col for col in runs.columns if runs[col].dtypes == 'O']
useless_vars = ['id', 'achievement_count', 'kudos_count', 'comment_count', 'pr_count']
tweak_runs = runs.drop(categorical_cols+useless_vars, axis=1)
tweak_runsNesse momento ocorrerá a separação das variáveis resposta e das que serão utilizadas para treinamento, de modo que após esta ação é possível realizar a aplicação do método SelectKBest com o uso de chi2 (ou outros testes estatísticos univariados) para a seleção de K feature mais importantes para o bom desempenho do modelo.
y = tweak_runs['pace']
X = tweak_runs.drop('pace',1)
best_features = SelectKBest(chi2, k=7).fit_transform(X, y.astype(int))
best_features
Outra alternativa para a seleção de features é com a aplicação de Recursive Feature Elimination (RFE), que treina o modelo selecionado e vai eliminando recursivamente as features de acordo com a falta de importância das mesmas. Mas como descrito, nesse ponto é necessário ter os modelos já selecionados para essa execução, então executaremos um para realizar a Regressão Linear e outro para a realização de Stochastic Gradient Descent (SGD), de modo que a variável resposta será diferente para esses dois modelos.
Vale dizer que o RFE é um método pesado para datasets de alta dimensionalidade, sendo assim uma alternativa é realizar o uso do SelectFromModel. Abaixo segue a definição de uma função para executar o RFE, bem como seu uso para um caso de regressão linear e SGD, respectivamente.
def get_best_rfe_features(X,y, model):
rfe = RFE(model, step=0.05).fit(X, y)
selected_features = [i for i, j in zip(X.columns, rfe.support_) if j]
return selected_featuresy = tweak_runs['pace']
X = tweak_runs.drop('pace',1)
encoded_y = preprocessing.LabelEncoder().fit_transform(y)
model = LinearRegression()
linear_feats = get_best_rfe_features(X, encoded_y, model)
y = tweak_runs['pace_sub_5']
X = tweak_runs.drop('pace_sub_5',1)
model = SGDClassifier(loss="hinge", penalty="l2", max_iter=5)
class_feats = get_best_rfe_features(X, y, model)
Outros método e combinações que podem ser aplicados, por exemplo, alterando a loss do modelo de SGD ou até mesmo a quantidade de steps do RFE, mas como o intuito desse artigo não é se aprofundar dessa forma, a apresentação dos métodos é suficiente para nos trazer bons resultados.
Aprendizado de Máquina
Dentre as técnicas de aprendizado de máquina, estaremos utilizando três nesse projeto, sendo para agrupamentos, regressão e classificação.
Agrupamentos
Dentre as técnicas de aprendizado de máquina que podem ser implementadas cita-se uma da esfera de aprendizado não supervisionado que é utilizada para realizar agrupamentos, a qual é denominada de K-means. Sendo assim, na prática essa técnica vai agrupar os registros de corridas que são semelhantes. Vale citar que esse método é estocástico, então cada execução poderá gerar resultados diferentes.
Para executar o K-means, basta separar a variável resposta das demais, na qual utilizaremos a função get_dummies do pandas para aproveitar também as variáveis categóricas, que serão transformadas em outras variáveis fictícias. Após a separação das variáveis basta escolher o número de clusters e passar para a classe K-means, juntamente com o fit no dataset de características. Para saber qual o cluster de cada amostra, realizaremos uma copia do dataset original de corrida e adicionaremos um identificador do cluster para os registros.
X = runs.drop('pace',1)
X = pd.get_dummies(X)
model = KMeans(n_clusters=4).fit(X)
clusterin_runs = runs.copy()
clusterin_runs['Cluster'] = model.labels_Feito isso podemos visualizar indicadores como a quantidade de registro em cada um dos clusters, média, desvio padrão, registros e outros, como demonstra os exemplos abaixo.
clustering_runs['Cluster'].value_counts()
clustering_runs.groupby('Cluster').mean()
clustering_runs.groupby('Cluster').std()
clustering_runs[clustering_runs['Cluster'] == 2]
Regressão
Quando se trata de regressões, os modelos são levados a prever um valor y dado um conjunto de características x, ou seja, o modelo aprende quais são os valores que compõem a equação da reta, ajustando várias linhas no mapa de características e retornando a linha com o menor erro entre os pontos.
Para a aplicação da regressão linear, iremos utilizar apenas as características que foram obtidas na seleção anterior e estão armazenadas na variável linear_feats . Além disso, iremos realizar a divisão do dataset em 80% para a fase de treino e 20% para a fase de teste, de modo que possamos metrificar a performance do algoritmo após a fase de treinamento.
y = runs['pace']
X = runs[linear_feats]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)Feita a divisão, basta submeter o dataset de treinamento para a função de fit da Regressão Linear e realizar predição em relação a base de teste.
model = LinearRegression()
model.fit(X_train,y_train)
y_pred = model.predict(X_test)Tendo os valores preditos e os valores reais, é possível calcular o Squared Mean Error (MSE, ou Erro Quadrático Médio) para demonstrar o quanto o modelo está errando, assim como é possível ver de forma gráfica o valor predito em relação ao valor real, por meio do código abaixo. O erro quadrático médio foi de 0.271 .
print('MSE:', metrics.mean_squared_error(y_test, y_pred))
plt.figure(figsize=(10,10))
plt.scatter(y_test, y_pred, c='crimson')
plt.yscale('log')
plt.xscale('log')
p1 = max(max(y_pred), max(y_test))
p2 = min(min(y_pred), min(y_test))
plt.plot([p1, p2], [p1, p2], 'b-')
plt.xlabel('True Values', fontsize=15)
plt.ylabel('Predictions', fontsize=15)
plt.axis('equal')
plt.show()
Para exemplificar, supondo que eu corra 5 km em 1488 segundos (24.8 minutos), a uma velocidade média de 4 m/s (14.4 km/h) e com velocidade maxima de 5.6 m/s (20.16 km/h) com a cadencia de 84 no ano de 2022, realizando um pace abaixo de 5, é previsto que meu pace seja igual a 4.1752 min/km .
model.predict(
pd.DataFrame(data={
'moving_time': 1488,
'average_speed': 4.0,
'max_speed': 5.6,
'average_cadence': 84.0,
'moving_time_minutes': 24.8,
'distance_km': 5.0,
'avg_speed_kmh': 14.4,
'max_speed_kmh': 20.16 ,
'year': 2022,
'pace_sub_5': True},
index=[0]
)
)Classificação
As classificações, por sua vez são utilizadas para diferenciar classes, ou seja, a variável resposta possui uma característica que rotula as amostras e dessa forma podemos buscar prever em qual rótulo uma determinada amostra receberá, baseado nas suas características.
Para realizar essa ação faremos algo semelhante ao que foi feito na etapa de regressão, separando as bases em 80% para treino e 20% para teste, bem como a divisão de amostras com as características recomendadas e variável resposta, que nesse caso será binária, representando se o pace da corrida foi abaixo de 5 ou não.
y = runs['pace_sub_5']
X = runs[class_feats]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)Tendo realizado a divisão das bases, iremos submeter ao fit do classificador SGDClassifier da biblioteca SKLearn, que implementa a regularização de modelos lineares com o uso de gradiente descendente estocástico para a otimização de parâmetros de forma iterativa alvitrando minimizar a função de interesse. No trecho de código abaixo também foi realizada a predição em relação ao dataset de testes.
model = SGDClassifier()
model.fit(X_train,y_train)
y_pred = model.predict(X_test)Diferente das regressões, para sistemas de classificações podemos utilizar outras métricas, como a precisão, que é razão entre as precisões corretas sobre o total de precisões realizadas, que no nosso caso foi de 95.12% . Para ter uma noção de verdadeiro positivo (valor real é positivo e previsão é positiva), verdadeiro negativo (valor real é negativo e previsão é negativa), falso positivo (valor real é negativo e previsão é positiva) e falso negativo (valor real é positivo e previsão é negativa), podemos desenvolver uma matriz de confusão, como demonstra o código abaixo.
print('Accuracy:', accuracy_score(y_test, y_pred))
cm = confusion_matrix(y_test, y_pred, labels=model.classes_)
disp = ConfusionMatrixDisplay(confusion_matrix=cm,display_labels=model.classes_)
disp.plot()
plt.show()
Sendo assim, com o intuito de exemplificar uma situação real, suponhamos que eu corra 5000 m com um tempo total de atividade igual a 1488 segundos (24.8min) e um tempo de movimento de 1440 segundos, com 25 m de ganho de elevação e uma diferença de 10 m de elevação (547 m de altitude máxima e 237 m de altitude mínima), alcançando 25 km/h de velocidade máxima e 210 bpm, eu conseguiria realizar a corrida com um pace médio abaixo de 5 min/km.
model.predict(
pd.DataFrame(data={
'distance': 5000 ,
'moving_time': 1440,
'elapsed_time': 1488,
'total_elevation_gain': 25,
'max_heartrate': 210,
'elev_high': 547,
'elev_low': 537,
'moving_time_minutes': 24.8,
'max_speed_kmh': 25,
'elev': 10
},
index=[0]
)
)Conclusão
As possibilidades de exploração de dados e a aplicação de diferentes tipos de modelos são incontáveis quando o dado se faz presente. Nesse artigo foi possível testar alguns deles e obter ótimos resultados em relação a insights e previsões para a corrida, sendo que um ponto a se considerar é que os dados estão baseado em um único atleta, ou seja, para que as conclusões fossem mais generalistas, seria necessário maior diversidade em relação aos atletas que compusessem o dataset, todavia se os passos iniciais forem seguidos, é possível que cada atleta consiga ter previsores e ideias relacionadas aos seus próprios dados.
Caso queira conferir os testes e o código utilizado, fique a vontade para acessar o seguinte repositório:
