I have the Python program below to download some earnings call transcripts through EDP-API. I have an enabled EDP-API and the program authenticates successfully, but it cannot find any earnings call transcripts. I am not sure if the transcripts are not accessible through the API, or if there is something wrong with the program. Any help would be appreciated.
Output:
INFO - Attempting authentication...
INFO - Authentication successful!
INFO - Fetching transcripts for AAPL.O from 2022-01-01 to 2022-12-31...
ERROR - Failed to get transcripts: 404 - 404 page not found
ERROR - No transcripts found or failed to download transcripts.
import requests
import base64
import json
from datetime import datetime, timedelta
import logging
import configparser
import os
# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class RDPClient:
def __init__(self, config_file='config.txt'):
"""Initialize client with credentials from config file"""
self.config = self._load_config(config_file)
self.api_key = self.config['api_key']
self.client_secret = self.config['client_secret']
self.username = self.config['username']
self.password = self.config['password']
self.base_url = "https://api.refinitiv.com"
self.auth_url = "https://api.refinitiv.com/auth/oauth2/v1/token"
self.access_token = None
self.token_expiry = None
def _load_config(self, config_file):
if not os.path.exists(config_file):
raise FileNotFoundError(f"Config file {config_file} not found.")
config = configparser.ConfigParser()
config.read(config_file)
try:
return {
'api_key': config['credentials']['api_key'],
'client_secret': config['credentials']['client_secret'],
'username': config['credentials']['username'],
'password': config['credentials']['password']
}
except KeyError as e:
raise KeyError(f"Missing required credential in config file: {e}")
def authenticate(self):
"""Authenticate with Refinitiv RDP API and get access token"""
try:
# Encode client credentials
credentials = f'{self.api_key}:{self.client_secret}'
encoded_credentials = base64.b64encode(credentials.encode()).decode()
headers = {
"Accept": "application/json",
"Content-Type": "application/x-www-form-urlencoded",
"Authorization": f"Basic {encoded_credentials}"
}
payload = {
"grant_type": "password",
"username": self.username,
"password": self.password,
"scope": "trapi",
"takeExclusiveSignOnControl": "true" # Forces other sessions to log out
}
logger.info("Attempting authentication...")
response = requests.post(self.auth_url, headers=headers, data=payload)
if response.status_code != 200:
error_data = response.json()
logger.error(f"Authentication failed: {error_data.get('error_description', 'Unknown error')}")
return False
auth_data = response.json()
self.access_token = auth_data['access_token']
expires_in = int(auth_data['expires_in'])
self.token_expiry = datetime.now() + timedelta(seconds=expires_in)
logger.info("Authentication successful!")
return True
except Exception as e:
logger.error(f"Authentication error: {str(e)}")
return False
def refresh_token(self):
"""Refresh token if expired"""
if datetime.now() >= self.token_expiry:
logger.info("Token expired, re-authenticating...")
return self.authenticate()
return True
def get_transcripts(self, symbol="AAPL.O", start_date="2022-01-01", end_date="2022-12-31"):
"""Fetch earnings call transcripts using RDP Filings API."""
if not self.refresh_token():
logger.error("Failed to refresh token.")
return None
try:
filings_url = f"{self.base_url}/data/filings/v1/graphql"
headers = {
"Authorization": f"Bearer {self.access_token}",
"Content-Type": "application/json"
}
# Construct the GraphQL query
query = {
"query": """
query ($symbol: String!, $startDate: String!, $endDate: String!) {
filings(filter: {ticker: {eq: $symbol}, filingDate: {gte: $startDate, lte: $endDate}}) {
edges {
node {
document {
content
title
filingDate
}
}
}
}
}
""",
"variables": {
"symbol": symbol,
"startDate": start_date,
"endDate": end_date
}
}
logger.info(f"Fetching transcripts for {symbol} from {start_date} to {end_date}...")
response = requests.post(filings_url, headers=headers, json=query)
if response.status_code == 200:
data = response.json()
if data.get('data') and data['data'].get('filings'):
filings = data['data']['filings']['edges']
transcripts = []
for filing in filings:
document = filing['node']['document']
if document and 'Earnings Call Transcript' in document['title']:
transcripts.append({
'title': document['title'],
'filingDate': document['filingDate'],
'content': document['content']
})
if transcripts:
logger.info(f"Found {len(transcripts)} earnings call transcripts.")
return transcripts
else:
logger.info("No earnings call transcripts found in the specified date range.")
return None
else:
logger.error(f"Failed to get transcripts: {response.status_code} - {response.text}")
return None
except Exception as e:
logger.error(f"Error getting transcripts: {str(e)}")
return None
def main():
try:
client = RDPClient('config.txt')
if client.authenticate():
transcripts = client.get_transcripts("AAPL.O", "2022-01-01", "2022-12-31")
if transcripts:
for i, transcript in enumerate(transcripts, start=1):
logger.info(f"Transcript {i}: {transcript['title']} on {transcript['filingDate']}")
logger.info(f"Content Preview: {transcript['content'][:500]}...") # Preview content
else:
logger.error("No transcripts found or failed to download transcripts.")
else:
logger.error("Authentication failed")
except Exception as e:
logger.error(f"An error occurred: {str(e)}")
if __name__ == "__main__":
main()
