Download data from Alphavantage http://www.alphavantage.co/ using Python, Requests and Pandas

alphavantage.py

import requests
from pandas.io.common import urlencode
from pandas.tseries.frequencies import to_offset


ALPHAVANTAGE_API_URL = "http://www.alphavantage.co/query"
ALPHAVANTAGE_API_KEY_DEFAULT = "demo"


def _init_session(session):
    if session is None:
        session = requests.Session()
    return session


def _url(url, params):
    if params is not None and len(params) > 0:
        return url + "?" + urlencode(params)
    else:
        return url


def get_ts_data(symbol, interval=None, outputsize=None, api_key=None, adjusted=False, session=None):
    session = _init_session(session)

    # apikey
    if api_key is None:
        api_key = ALPHAVANTAGE_API_KEY_DEFAULT

    # function
    d_functions = {
        to_offset("D").freqstr: "TIME_SERIES_DAILY",
        to_offset("W").freqstr: "TIME_SERIES_WEEKLY",
        to_offset("M").freqstr: "TIME_SERIES_MONTHLY",
    }
    try:
        if adjusted and to_offset(interval).freqstr == "D":
            function_api = "TIME_SERIES_DAILY_ADJUSTED"
        else:
            function_api = d_functions[to_offset(interval).freqstr]
    except KeyError:
        function_api = "TIME_SERIES_INTRADAY"

    # interval
    if interval is None:
        interval = "D"
    d_intervals = {
        to_offset("1T").freqstr: "1min",
        to_offset("5T").freqstr: "5min",
        to_offset("15T").freqstr: "15min",
        to_offset("30T").freqstr: "30min",
        to_offset("H").freqstr: "60min",
        to_offset("D").freqstr: "daily",
        to_offset("W").freqstr: "weekly",
        to_offset("M").freqstr: "monthly"
    }
    try:
        interval_api = d_intervals[to_offset(interval).freqstr]
    except KeyError:
        interval_api = "60min"

    # outputsize
    if outputsize is None:
        outputsize = "compact"

    params = {
        "function": function_api,
        "symbol": symbol,
        "interval": interval_api,
        "outputsize": outputsize,
        "apikey": api_key
    }

    r = session.get(ALPHAVANTAGE_API_URL, params=params)
    url_long= _url(ALPHAVANTAGE_API_URL, params)
    # print(url_long)
    if r.status_code == requests.codes.ok:
        dat = r.json()
        metadata = dat["Meta Data"]
        key_dat = list(dat.keys())[1]  # ugly
        ts = dat[key_dat]
        df = pd.DataFrame(ts).T
        df = df.rename(columns={
            "1. open": "Open",
            "2. high": "High",
            "3. low": "Low",
            "4. close": "Close",
            "5. volume": "Volume",
        })
        for col in ["Open", "High", "Low", "Close"]:
            if col in df.columns:
                df[col] = df[col].astype(float)
        df["Volume"] = df["Volume"].astype(int)
        df.index = pd.to_datetime(df.index)
        df.index.name = "Date"
        return df, metadata
    else:
        params["apikey"] = "HIDDEN"
        raise Exception(r.status_code, r.reason, url_long)


def get_sector_performances(api_key=None, session=None):
    session = _init_session(session)

    # apikey
    if api_key is None:
        api_key = ALPHAVANTAGE_API_KEY_DEFAULT

    params = {
        "function": "SECTOR",
        "apikey": api_key
    }
    url_long = _url(ALPHAVANTAGE_API_URL, params)
    print(url_long)
    r = session.get(ALPHAVANTAGE_API_URL, params=params)
    if r.status_code == requests.codes.ok:
        dat = r.json()
        metadata = dat["Meta Data"]
        del dat["Meta Data"]
        df = pd.DataFrame(dat)
        for col in df.columns:
            df[col] = df[col].str.strip("%").astype(float)
        return df, metadata
    else:
        params["apikey"] = "HIDDEN"
        raise Exception(r.status_code, r.reason, url_long)
    

import os
import pandas as pd
pd.set_option("max_rows", 10)
import datetime
import requests_cache

expire_after = datetime.timedelta(days=1)
session = requests_cache.CachedSession(cache_name='cache', backend='sqlite', expire_after=expire_after)

api_key = os.environ.get("ALPHAVANTAGE_API_KEY")  # api_key = "YOURAPIKEY"
df, metadata = get_ts_data("MSFT", interval="W", api_key=api_key, session=session)
print(df)
print(df.dtypes)
print(metadata)

df, metadata = get_ts_data("MSFT", interval="15Min", api_key=api_key, session=session)
print(df)
print(df.dtypes)
print(metadata)

df, metadata = get_ts_data("MSFT", interval="D", api_key=api_key, session=session)
print(df)
print(df.dtypes)
print(metadata)

df, metadata = get_ts_data("MSFT", interval="D", outputsize="full", api_key=api_key, session=session)
print(df)
print(df.dtypes)
print(metadata)

df, metadata = get_sector_performances(api_key=api_key, session=session)
print(df)
print(df.dtypes)
print(metadata)

From pydata/pandas-datareader#315

export ALPHAVANTAGE_API_KEY = "YOURAPIKEY"

ToDo: create a GitHub project (similar to https://github.com/femtotrader/python-eodhistoricaldata )

CSV output is now available https://www.alphavantage.co/documentation/

Optional: datatype

By default, datatype=json. Strings json and csv are accepted

When i tried to run this code shows this message D:\Code> python alphavantage.py Traceback (most recent call last): File "alphavantage.py", line 133, in <module> import requests_cache ModuleNotFoundError: No module named 'requests_cache'