Get Notified when a Public Company cuts their Dividend using Python + Polygon.io

We are going to write a basic script that detects when a company cuts a dividend. We will be using Python and Polygon.io for financial data. You will need an account (Free to sign up).

Polygon.io offers a Python client, we won’t be using it. We will just call REST API. For our script, we will take two arguments:

• Ticker (required)
• After Date (optional)

The Python call will look like:

POLYGON_API_KEY="YOUR_SECRET_KEY" python3 main.py AAPL 2020-01-01

or,

POLYGON_API_KEY="YOUR_SECRET_KEY" python3 main.py AAPL

Now, the script (below, split into parts and explained):

import os
import sys
import time
from datetime import datetime
import requests

DATE_FORMAT = '%Y-%m-%d'
DEFAULT_INIT_DATE = '1980-01-01'

# get_all_dividends :: This will fetch a list of past dividends from the Polygon API.
# It will return a list with the following fields:
#  - cash_amount
#  - ex_dividend_date
# 
def get_all_dividends(ticker, initial_date):
    previous_minute = -1
    current_count = 0

    results = []
    url = build_url(ticker, initial_date)
    while True:
        current_minute = datetime.now().minute
        if current_minute != previous_minute:
            # Reset the rate limit counter
            previous_minute = current_minute
            current_count = 0
        
        if current_count >= 5:
            # Wait a minute
            print ('Waiting 10 seconds, then seeing if we can continue...')
            time.sleep(10)
            continue
        
        current_count += 1 # Increase the rate limit counter
        page_of_results, next_url, rate_limited = get_single_page_of_results(url)
        if rate_limited:
            current_count = 5 # Force a wait
            time.sleep(1)
            continue
        
        url = next_url
        results += page_of_results
        if not next_url:
            break
    return results

# Build the initial url
def build_url(ticker, initial_date):
    url = 'https://api.polygon.io/v3/reference/dividends?ticker=' + ticker
    url += '&apiKey=' + get_polygon_api_key()
    url += '&limit=1000' # Using 1000 because its the max
    url += '&ex_dividend_date.gte=' + initial_date.strftime(DATE_FORMAT)
    url += '&sort=ex_dividend_date'
    url += '&order=asc'
    return url

def get_single_page_of_results(url):
    if url == '':
        # The next_url came back empty, so we are done
        return [], '', False
    
    # Fetch the data
    r = requests.get(url, headers={'Authorization': 'Bearer ' + get_polygon_api_key()}, timeout=30)
    if r.status_code == 429:
        # We were rate limited
        return [], '', True
    if r.status_code != 200:
        # The request has failed, and we should stop the script
        print ('Error: ' + str(r.status_code) + ' - ' + r.text)
        sys.exit(1)
    
    # Extract the response
    data = r.json()
    next_url = ''
    if 'next_url' in data:
        next_url = data['next_url']

    return data.get('results', []), next_url, False

def get_polygon_api_key():
    if not os.environ.get('POLYGON_API_KEY'):
        print ('Please set the POLYGON_API_KEY environment variable')
        sys.exit(1)
    return os.environ['POLYGON_API_KEY']

def main():
    # Get the arguments
    if len(sys.argv) < 2:
        print ('Please provide a ticker symbol')
        return
    ticker = sys.argv[1]
    ticker = ticker.upper()
    print ('Ticker: ' + ticker)

    # Get the second argument if it exists
    initial_date = datetime.strptime(DEFAULT_INIT_DATE, DATE_FORMAT).date()
    if len(sys.argv) > 2:
        initial_date = datetime.strptime(sys.argv[2], DATE_FORMAT).date()
    print ('Initial Date: ' + str(initial_date))

    all_dividends = get_all_dividends(ticker, initial_date)
    # All Dividends is a list of objects. An object contains the following
    # fields:
    # - cash_amount
    # - declaration_date
    # - dividend_type
    # - ex_dividend_date
    # - frequency
    # - pay_date
    # - record_date
    previous_cash_amount = 0
    cuts = 0
    for dividend in all_dividends:
        if dividend['dividend_type'] != 'CD':
            # Not a consistent dividend
            continue
        
        if dividend['cash_amount'] < previous_cash_amount:
            # This dividend is less than the previous one
            print ('Dividend cut detected: ' + str(dividend['ex_dividend_date']))
            cuts += 1
        previous_cash_amount = dividend['cash_amount']
    
    print ('Done, total cuts: ' + str(cuts))

main()

Above is the entire script, and below I’ve split it into parts. Let’s start with the imports:

import os
import sys
import time
from datetime import datetime
import requests

We use the datetime library to parse the dates to strings and back. We use the sys for arguments, and exiting with error status codes. We use requests to make the HTTP requests. datetime, os, sys and time are all standard libraries. You may need to run the following command for requests.

python3 -m pip install requests

Next, we have two constants:

DATE_FORMAT = '%Y-%m-%d'
DEFAULT_INIT_DATE = '1980-01-01'

As mentioned, the second optional argument is the initial date. If not provided, we go as far as January 1, 1980. I did this because it decreases the amount of data being pulled. I think 40 years is enough. The DATE_FORMAT is the expected format for the date.

Next, let’s look at the code to make the request against Polygon. We are calling the dividends endpoint.

# get_all_dividends :: This will fetch a list of past dividends from the Polygon API.
# It will return a list with the following fields:
#  - cash_amount
#  - ex_dividend_date
# 
def get_all_dividends(ticker, initial_date):
    previous_minute = -1
    current_count = 0

    results = []
    url = build_url(ticker, initial_date)
    while True:
        current_minute = datetime.now().minute
        if current_minute != previous_minute:
            # Reset the rate limit counter
            previous_minute = current_minute
            current_count = 0
        
        if current_count >= 5:
            # Wait a minute
            print ('Waiting 10 seconds, then seeing if we can continue...')
            time.sleep(10)
            continue
        
        current_count += 1 # Increase the rate limit counter
        page_of_results, next_url, rate_limited = get_single_page_of_results(url)
        if rate_limited:
            current_count = 5 # Force a wait
            time.sleep(1)
            continue
        
        url = next_url
        results += page_of_results
        if not next_url:
            break
    return results

# Build the initial url
def build_url(ticker, initial_date):
    url = 'https://api.polygon.io/v3/reference/dividends?ticker=' + ticker
    url += '&apiKey=' + get_polygon_api_key()
    url += '&limit=1000' # Using 1000 because its the max
    url += '&ex_dividend_date.gte=' + initial_date.strftime(DATE_FORMAT)
    url += '&sort=ex_dividend_date'
    url += '&order=asc'
    return url

def get_single_page_of_results(url):
    if url == '':
        # The next_url came back empty, so we are done
        return [], '', False
    
    # Fetch the data
    r = requests.get(url, headers={'Authorization': 'Bearer ' + get_polygon_api_key()}, timeout=30)
    if r.status_code == 429:
        # We were rate limited
        return [], '', True
    if r.status_code != 200:
        # The request has failed, and we should stop the script
        print ('Error: ' + str(r.status_code) + ' - ' + r.text)
        sys.exit(1)
    
    # Extract the response
    data = r.json()
    next_url = ''
    if 'next_url' in data:
        next_url = data['next_url']

    return data.get('results', []), next_url, False

def get_polygon_api_key():
    if not os.environ.get('POLYGON_API_KEY'):
        print ('Please set the POLYGON_API_KEY environment variable')
        sys.exit(1)
    return os.environ['POLYGON_API_KEY']

The first aspect of this is the rate limiting. I have a free account and I’m limited to 5 requests per minute. The code starts by checking if the current minute is different than the one in previous_minute. If so, it resets the count. Otherwise, we check the count. If count is greater than or equal to 5, wait 10 seconds. This could be less, but most companies will require one call.

We then make the request. We get three arguments back, the results (list of dividend data), the next url if there’s more pages, or we were rate limited (and need to wait).

Why the rate limit check again? Technically, we could use our rate limit in multiple spots and be throttled. As of right now, this code could scale horizontally across multiple computers, but each process tracks its own request count. The API can still throttle requests made with the same key.

We move into the next function: build_url . This is only called for the initial url. Polygon returns a full URL in the response to use in the next API call. I determined the majority of these values by looking at the API docs. The sort matters because we are looking if dividends have been cut (reversing would mess up the logic).

Finally, get_single_page_of_results function. This is making an API call, extracting some response values or checking if the HTTP status code returned a non-200.

This would set us up with all dividend information. Now, let’s look at the main function.

def main():
    # Get the arguments
    if len(sys.argv) < 2:
        print ('Please provide a ticker symbol')
        return
    ticker = sys.argv[1]
    ticker = ticker.upper()
    print ('Ticker: ' + ticker)

    # Get the second argument if it exists
    initial_date = datetime.strptime(DEFAULT_INIT_DATE, DATE_FORMAT).date()
    if len(sys.argv) > 2:
        initial_date = datetime.strptime(sys.argv[2], DATE_FORMAT).date()
    print ('Initial Date: ' + str(initial_date))

    all_dividends = get_all_dividends(ticker, initial_date)
    
    # All Dividends is a list of objects. An object contains the following
    # fields:
    # - cash_amount
    # - declaration_date
    # - dividend_type
    # - ex_dividend_date
    # - frequency
    # - pay_date
    # - record_date
    previous_cash_amount = 0
    cuts = 0
    for dividend in all_dividends:
        if dividend['dividend_type'] != 'CD':
            # Not a consistent dividend
            continue
        
        if dividend['cash_amount'] < previous_cash_amount:
            # This dividend is less than the previous one
            print ('Dividend cut detected: ' + str(dividend['ex_dividend_date']))
            cuts += 1
        previous_cash_amount = dividend['cash_amount']
    
    print ('Done, total cuts: ' + str(cuts))

As you can see, we just iterate through the dividends checking for a cut on the cash amount.

Below is a screenshot of me running the script:

In terms of notifying you, I won’t demonstrate that part. You can setup a simple Slackbot via webhook, or email with Sendgrid.

Thanks for reading! You can make this code much more efficient, but I chose to keep it simple.