financial-machine-learning/git_status.py

import os
from typing import Dict
import datetime
from conf import PROJECT_ROOT_DIR
import re
import pandas as pd
from github import Github, Repository


def get_github_client():
    # search for app_client and client secrets first, since this allow higher api request limit
    github_app = os.environ.get('GIT_APP_ID')
    if github_app is None:
        github_token = os.environ.get('GIT_TOKEN')
        g = Github(github_token)
    else:
        github_app_secret = os.environ.get('GIT_APP_SECRET')
        g = Github(
            client_id=github_app,
            client_secret=github_app_secret)
    return g


# generic search functions
def search_repo(search_term: str, qualifier_dict: Dict):
    g = get_github_client()
    qualifier_str = ' '.join(['{}:{}'.format(k, v) for k, v in iter(qualifier_dict.items())])
    if qualifier_str != '':
        final_search_term = '{} {}'.format(search_term, qualifier_str)
    else:
        final_search_term = search_term
    repo_result = g.search_repositories(final_search_term)
    return repo_result


def search_repo_simple(search_term: str = None,
                       min_stars_number: int = None,
                       created_at: str = None,
                       pushed_date: str = None
                       ):
    """

    :param search_term:
    :param min_stars_number:
    :param created_at:
    :param pushed_date:
    usage:
    >>> search_term = '(deep learning) AND trading'
    >>> min_stars_number = 100
    >>> created_at = None
    >>> pushed_date = None
    """
    if search_term is None:
        _search_term = ''
    else:
        _search_term = search_term

    qualifier_dict = {}
    if min_stars_number is not None:
        qualifier_dict['stars'] = '>={}'.format(min_stars_number)

    if created_at is not None:
        qualifier_dict['created'] = '>={}'.format(created_at)

    if pushed_date is not None:
        qualifier_dict['pushed'] = '>={}'.format(pushed_date)
    search_result = search_repo(_search_term, qualifier_dict)
    return search_result


# *******
# topic specific search functions
# *******
def convert_repo_list_to_df(repo_list, category):
    df_list = []
    for repo in repo_list:
        attr_dict = get_repo_attributes_dict(repo)
        attr_dict['name'] = repo.name
        attr_dict['comment'] = 'NEW'
        attr_dict['category'] = category
        attr_dict['repo_path'] = repo.full_name
        attr_dict['url'] = 'https://github.com/{}'.format(repo.full_name)
        df_list.append(attr_dict)
    result_df = pd.DataFrame(df_list)
    return result_df


def search_new_repo_and_append(min_stars_number: int = 100):
    repo_df = get_repo_list()
    category_list = repo_df['category'].unique().tolist()
    new_repo_list = []
    for category in category_list:
        if category == 'Deep Learning':
            # github not yet support OR operator, issue here
            # https://github.com/isaacs/github/issues/660
            # hence run the search terms twice and combine
            search_term = 'deep learning trading'
            repo_list = search_repo_simple(search_term, min_stars_number)
            top_df = convert_repo_list_to_df(repo_list, category)
            search_term = 'deep learning finance'
            repo_list = search_repo_simple(search_term, min_stars_number)
            bottom_df = convert_repo_list_to_df(repo_list, category)
            combined_df = pd.concat([top_df, bottom_df]).reset_index(drop=True)
            combined_df = combined_df.drop_duplicates()
            # only find ones that need to be inserted
            combined_df = combined_df[~combined_df['repo_path'].str.lower().isin(repo_df['repo_path'].str.lower())]
            combined_df['finml_added_date'] = datetime.datetime.now()
            new_repo_list.append(combined_df)
    new_repo_df = pd.concat(new_repo_list).reset_index(drop=True)
    final_df = pd.concat([repo_df, new_repo_df]).reset_index(drop=True)
    final_df = final_df.sort_values(by='category')
    final_df.to_csv(os.path.join(PROJECT_ROOT_DIR, 'raw_data', 'url_list.csv'), index=False)


# *******
# saved repo list, treat it as database for now
# *******
def get_repo_list():
    repo_df = pd.read_csv(os.path.join(PROJECT_ROOT_DIR, 'raw_data', 'url_list.csv'))
    if 'repo_path' not in repo_df.columns:
        repo_df['repo_path'] = repo_df['url'].apply(get_repo_path)
    return repo_df


# *******
# repo specific information
# *******
def get_repo_path(in_url):
    repo_path = None
    if 'https://github.com/' in in_url:
        url_query = in_url.replace('https://github.com/', '')
        repo_path = '/'.join(url_query.split('/')[:2])
    return repo_path


def get_last_commit_date(input_repo: Repository):
    """
    get latest commit from repo
    :param input_repo:
    :return:
    """
    page = input_repo.get_commits().get_page(0)[0]
    return page.commit.author.date


def get_repo_attributes_dict(input_repo: Repository, last_commit_within_years: int = 2):
    result_dict = {
        'repo_path': input_repo.full_name,
        'created_at': input_repo.created_at,
        'last_commit': get_last_commit_date(input_repo),
        'last_update': input_repo.updated_at,
        'star_count': input_repo.stargazers_count,
        'fork_count': input_repo.forks_count,
        'contributors_count': input_repo.get_contributors().totalCount

    }
    today = datetime.datetime.today()
    check_start_date = datetime.datetime(today.year - last_commit_within_years,
                                         today.month,
                                         today.day)

    if result_dict['last_commit'] >= check_start_date:
        repo_status = 'active'
    else:
        repo_status = 'inactive'
    result_dict['repo_status'] = repo_status

    return result_dict


def get_repo_status():
    g = get_github_client()
    repo_df = get_repo_list()
    for idx, row in repo_df.iterrows():
        repo_path = row['repo_path']
        if repo_path is not None:
            try:
                print('processing [{}]'.format(repo_path))
                repo = g.get_repo(repo_path)
                repo_attr_dict = get_repo_attributes_dict(repo)
            except Exception as ex:
                print(ex)
                repo_attr_dict = {}

            for k, v in iter(repo_attr_dict.items()):
                repo_df.loc[idx, k] = v
    repo_df.to_csv(os.path.join(PROJECT_ROOT_DIR, 'raw_data', 'url_list.csv'), index=False)


@DeprecationWarning
def parse_readme_md():
    """

    :return:
    usage:
    >>> df = parse_readme_md()
    >>> df.to_csv(os.path.join(PROJECT_ROOT_DIR, 'raw_data', 'url_list.csv'), index=False)
    """
    file_path = os.path.join(PROJECT_ROOT_DIR, 'README.md')
    with open(file_path) as f:
        lines = f.readlines()[11:]  # skip heading
        all_df_list = []
        for line_num in range(len(lines)):
            line = lines[line_num]
            if line.strip().startswith('#'):
                # find a heading
                heading = line.strip().replace('#', '').replace('\n', '').strip()
                # parse until next # or eof
                parsed_list = []
                line_num += 1
                while line_num < len(lines) and not lines[line_num].strip().startswith('#'):
                    link_line = lines[line_num].replace('\n', '').strip()
                    if len(link_line) > 0:
                        # usually in the format of '- [NAME](link) - comment
                        split_sections = link_line.split('- ')
                        if len(split_sections) == 2:
                            comment_str = None
                        elif len(split_sections) >= 3:
                            comment_str = '-'.join(split_sections[2:]).strip()
                        else:
                            raise Exception('link_line [{}] not supported'.format(link_line))

                        title_and_link = split_sections[1].strip()
                        title = re.search(r'\[(.*?)\]', title_and_link)
                        title_str = None
                        if title is not None:
                            title_str = title.group(1)
                            title_and_link = title_and_link.replace('[{}]'.format(title_str), '')
                        m_link = re.search(r'\((.*?)\)', title_and_link)
                        link_str = None
                        if m_link is not None:
                            link_str = m_link.group(1)
                        parsed_set = (title_str, link_str, comment_str)
                        parsed_list.append(parsed_set)
                    line_num += 1
                parsed_df = pd.DataFrame(parsed_list, columns=['name', 'url', 'comment'])
                parsed_df['category'] = heading
                all_df_list.append(parsed_df)
    final_df = pd.concat(all_df_list).reset_index(drop=True)
    return final_df


if __name__ == '__main__':
    get_repo_status()
    search_new_repo_and_append(min_stars_number=100)