Case Study 21-1: Priya Builds a Market Intelligence Dashboard
The Situation
It is a Thursday afternoon in late October. Priya Okonkwo has just come out of a meeting with Sandra Chen and she has a problem.
The meeting was about Q4 strategy. Sandra's main concern: Acme Corp's biggest regional competitor, Pinnacle Office Solutions, has been aggressively cutting prices on commercial paper and toner cartridges — Acme's two highest-volume SKUs. Two mid-market accounts have mentioned Pinnacle's pricing in recent renewal conversations. One defected last quarter.
Sandra needs to understand the competitive landscape. Are Pinnacle's cuts isolated or part of a broader trend? Is the wider office supplies distribution market contracting? What are the economic signals — inflation, currency movements, supply chain news — that might be influencing pricing?
"I can't make a Q4 pricing recommendation without more context," Sandra told her. "I need to know what's happening in the market, not just what's happening in our spreadsheets."
Priya has been working through the API chapter of her Python course. She decides to build something that combines external data with Acme's internal numbers.
What Priya Builds
Her plan is a weekly market intelligence pull that:
- Fetches recent news headlines about the office supplies industry and their key competitors
- Pulls current currency exchange rates (Acme's Canadian account is increasingly significant)
- Retrieves basic financial data on Pinnacle Office Solutions' parent company (publicly traded)
- Combines these with Acme's own internal weekly sales data
- Outputs a summary report in a format Sandra can actually use
The Code
Step 1: Environment Setup
Priya creates a .env file in her project directory:
# .env — never commit this file
NEWS_API_KEY=your_newsapi_key_here
ALPHA_VANTAGE_API_KEY=your_alphavantage_key_here
And a requirements.txt:
requests==2.31.0
python-dotenv==1.0.0
Step 2: News Intelligence
# market_intelligence.py
import os
import csv
import logging
from datetime import datetime, timedelta
from dotenv import load_dotenv
import requests
load_dotenv()
logger = logging.getLogger(__name__)
def fetch_industry_news(search_queries: list[str], days_back: int = 7) -> list[dict]:
"""
Fetch recent news for multiple search queries.
Deduplicates by URL so overlapping queries don't return the same article twice.
Args:
search_queries: List of search strings to query
days_back: How many days of news to retrieve
Returns:
List of unique article dicts sorted by publish date descending
"""
api_key = os.environ.get("NEWS_API_KEY")
if not api_key:
raise ValueError(
"NEWS_API_KEY not found. Set it in your .env file. "
"Get a free key at https://newsapi.org/register"
)
from_date = (datetime.now() - timedelta(days=days_back)).strftime("%Y-%m-%d")
seen_urls = set()
all_articles = []
for query in search_queries:
logger.info(f"Fetching news for query: '{query}'")
try:
response = requests.get(
"https://newsapi.org/v2/everything",
headers={"X-Api-Key": api_key},
params={
"q": query,
"from": from_date,
"language": "en",
"sortBy": "relevancy",
"pageSize": 15,
},
timeout=20,
)
response.raise_for_status()
data = response.json()
if data.get("status") != "ok":
logger.warning(
f"News API error for query '{query}': {data.get('message')}"
)
continue
for article in data.get("articles", []):
url = article.get("url", "")
if url and url not in seen_urls:
seen_urls.add(url)
all_articles.append({
"search_query": query,
"title": article.get("title", ""),
"source": article.get("source", {}).get("name", ""),
"published_date": article.get("publishedAt", "")[:10],
"description": article.get("description", "")[:300],
"url": url,
})
except requests.exceptions.RequestException as e:
logger.error(f"Failed to fetch news for '{query}': {e}")
continue
# Sort by publish date, most recent first
all_articles.sort(key=lambda a: a["published_date"], reverse=True)
logger.info(f"Retrieved {len(all_articles)} unique articles across {len(search_queries)} queries")
return all_articles
Step 3: Exchange Rates for Canadian Account Analysis
def fetch_usd_cad_rate() -> dict:
"""
Fetch the current USD/CAD exchange rate.
Used to normalize Acme's Canadian account revenue for comparison.
Returns:
Dict with rate, direction, and timestamp
"""
logger.info("Fetching USD/CAD exchange rate...")
response = requests.get(
"https://open.er-api.com/v6/latest/USD",
timeout=15,
)
response.raise_for_status()
data = response.json()
if data.get("result") != "success":
raise ValueError(f"Exchange rate API error: {data}")
rates = data["rates"]
cad_rate = rates.get("CAD", 1.0)
return {
"usd_to_cad": cad_rate,
"cad_to_usd": round(1 / cad_rate, 6),
"last_updated": data.get("time_last_update_utc", ""),
}
Step 4: Competitor Financial Context
def fetch_competitor_financials(ticker_symbol: str) -> dict:
"""
Fetch public company financials from Alpha Vantage.
Used to understand the parent company of Acme's regional competitor.
Free tier: 25 requests/day, 5 requests/minute.
Get a free key at https://www.alphavantage.co/support/#api-key
"""
api_key = os.environ.get("ALPHA_VANTAGE_API_KEY")
if not api_key:
logger.warning(
"ALPHA_VANTAGE_API_KEY not set — skipping competitor financials"
)
return {}
logger.info(f"Fetching financials for {ticker_symbol}...")
try:
response = requests.get(
"https://www.alphavantage.co/query",
params={
"function": "OVERVIEW",
"symbol": ticker_symbol,
"apikey": api_key,
},
timeout=20,
)
response.raise_for_status()
data = response.json()
if "Note" in data:
# Alpha Vantage rate limit message
logger.warning(
"Alpha Vantage API rate limit hit. "
"Wait 1 minute or upgrade your API plan."
)
return {}
if not data or "Name" not in data:
logger.warning(f"No financial data found for ticker: {ticker_symbol}")
return {}
return {
"ticker": ticker_symbol,
"company_name": data.get("Name"),
"industry": data.get("Industry"),
"sector": data.get("Sector"),
"description": data.get("Description", "")[:250],
"market_cap": data.get("MarketCapitalization"),
"revenue_ttm": data.get("RevenueTTM"),
"profit_margin": data.get("ProfitMargin"),
"pe_ratio": data.get("PERatio"),
"52_week_high": data.get("52WeekHigh"),
"52_week_low": data.get("52WeekLow"),
"analyst_target_price": data.get("AnalystTargetPrice"),
}
except requests.exceptions.RequestException as e:
logger.error(f"Failed to fetch financials for {ticker_symbol}: {e}")
return {}
Step 5: Loading Acme's Internal Sales Data
def load_acme_weekly_sales(csv_path: str) -> dict:
"""
Load Acme's internal weekly sales summary from a CSV file.
Expected columns: region, product_category, weekly_revenue, units_sold, week_ending
Returns:
Dict with totals and region breakdown
"""
weekly_totals = {}
regional_totals = {}
total_revenue = 0.0
total_units = 0
with open(csv_path, "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
region = row.get("region", "Unknown")
category = row.get("product_category", "Unknown")
revenue = float(row.get("weekly_revenue", 0))
units = int(row.get("units_sold", 0))
# Accumulate by region
if region not in regional_totals:
regional_totals[region] = {"revenue": 0.0, "units": 0}
regional_totals[region]["revenue"] += revenue
regional_totals[region]["units"] += units
# Accumulate by product category
if category not in weekly_totals:
weekly_totals[category] = {"revenue": 0.0, "units": 0}
weekly_totals[category]["revenue"] += revenue
weekly_totals[category]["units"] += units
total_revenue += revenue
total_units += units
return {
"total_revenue": total_revenue,
"total_units": total_units,
"by_region": regional_totals,
"by_category": weekly_totals,
}
Step 6: Assembling the Report
def generate_market_intelligence_report(
news_articles: list[dict],
exchange_data: dict,
competitor_data: dict,
acme_sales: dict,
output_path: str,
) -> None:
"""
Write a weekly market intelligence brief to a text file.
Formatted for easy reading in Sandra Chen's email.
"""
today = datetime.now().strftime("%Y-%m-%d")
with open(output_path, "w", encoding="utf-8") as report:
report.write(
f"ACME CORP — WEEKLY MARKET INTELLIGENCE BRIEF\n"
f"Prepared by: Priya Okonkwo, Analytics\n"
f"Date: {today}\n"
f"{'=' * 65}\n\n"
)
# Section 1: Acme Internal Summary
report.write("1. ACME INTERNAL PERFORMANCE (This Week)\n")
report.write("-" * 45 + "\n")
report.write(
f" Total Revenue: ${acme_sales.get('total_revenue', 0):>12,.2f}\n"
f" Total Units Sold: {acme_sales.get('total_units', 0):>12,}\n\n"
)
report.write(" By Region:\n")
for region, data in sorted(acme_sales.get("by_region", {}).items()):
report.write(
f" {region:<20} ${data['revenue']:>12,.2f} "
f"({data['units']:,} units)\n"
)
report.write("\n By Product Category:\n")
categories_sorted = sorted(
acme_sales.get("by_category", {}).items(),
key=lambda x: x[1]["revenue"],
reverse=True,
)
for category, data in categories_sorted:
pct = (data["revenue"] / acme_sales["total_revenue"] * 100
if acme_sales["total_revenue"] else 0)
report.write(
f" {category:<25} ${data['revenue']:>10,.2f} ({pct:.1f}%)\n"
)
# Section 2: Currency context
report.write("\n2. CURRENCY CONTEXT\n")
report.write("-" * 45 + "\n")
if exchange_data:
usd_to_cad = exchange_data.get("usd_to_cad", "N/A")
cad_to_usd = exchange_data.get("cad_to_usd", "N/A")
report.write(
f" USD/CAD Rate: 1 USD = {usd_to_cad:.4f} CAD\n"
f" 1 CAD = {cad_to_usd:.4f} USD\n"
f" Last updated: {exchange_data.get('last_updated', 'unknown')}\n"
)
# Canadian account revenue in USD
canadian_revenue_cad = (
acme_sales.get("by_region", {})
.get("Canada", {})
.get("revenue", 0)
)
if canadian_revenue_cad:
canadian_revenue_usd = canadian_revenue_cad / usd_to_cad
report.write(
f"\n Canadian region revenue at current rates:\n"
f" CAD {canadian_revenue_cad:,.2f} = USD {canadian_revenue_usd:,.2f}\n"
)
# Section 3: Competitor Context
report.write("\n3. COMPETITOR FINANCIAL CONTEXT\n")
report.write("-" * 45 + "\n")
if competitor_data:
report.write(
f" Company: {competitor_data.get('company_name', 'N/A')}\n"
f" Ticker: {competitor_data.get('ticker', 'N/A')}\n"
f" Industry: {competitor_data.get('industry', 'N/A')}\n"
f" Market Cap: ${float(competitor_data.get('market_cap', 0)):,.0f}\n"
f" Revenue (TTM): ${float(competitor_data.get('revenue_ttm', 0)):,.0f}\n"
f" Profit Margin: {competitor_data.get('profit_margin', 'N/A')}\n"
f" P/E Ratio: {competitor_data.get('pe_ratio', 'N/A')}\n"
f" 52-Week Range: "
f"{competitor_data.get('52_week_low', '?')} – "
f"{competitor_data.get('52_week_high', '?')}\n"
)
if competitor_data.get("description"):
report.write(f"\n Business Description:\n")
desc_words = competitor_data["description"].split()
line = " "
for word in desc_words:
if len(line) + len(word) + 1 > 65:
report.write(line + "\n")
line = " " + word + " "
else:
line += word + " "
if line.strip():
report.write(line + "\n")
else:
report.write(
" [Set ALPHA_VANTAGE_API_KEY to enable competitor financials]\n"
)
# Section 4: Industry News
report.write(f"\n4. INDUSTRY NEWS ({len(news_articles)} articles, last 7 days)\n")
report.write("-" * 45 + "\n")
if news_articles:
for article in news_articles[:10]:
report.write(
f"\n [{article['published_date']}] {article['source']}\n"
f" {article['title']}\n"
)
if article.get("description"):
report.write(
f" {article['description'][:150]}...\n"
)
report.write(f" {article['url']}\n")
else:
report.write(
" [Set NEWS_API_KEY to enable industry news monitoring]\n"
)
report.write(f"\n{'=' * 65}\n")
report.write("End of Market Intelligence Brief\n")
report.write(
"Data sources: NewsAPI, ExchangeRate-API, Alpha Vantage, Acme internal CSV\n"
)
logger.info(f"Market intelligence report saved to {output_path}")
print(f"Report written to: {output_path}")
Step 7: The Main Runner
def main():
"""Run the full market intelligence pipeline."""
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)s %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
today = datetime.now().strftime("%Y-%m-%d")
output_path = f"acme_market_brief_{today}.txt"
# Search queries for industry monitoring
news_queries = [
"office supplies distribution wholesale",
"Pinnacle Office Solutions", # Direct competitor monitoring
"commercial paper toner supply chain",
"office products market pricing",
]
print("Pulling market intelligence data...")
# Fetch all external data
news_articles = fetch_industry_news(news_queries, days_back=7)
exchange_data = fetch_usd_cad_rate()
# Note: replace "SPLS" with the actual parent company ticker
# Staples parent company (example — use the ticker relevant to your competitor)
competitor_data = fetch_competitor_financials("SPLS")
# Load Acme's internal sales (use a real file path in production)
# For demo purposes, we use a sample structure
acme_sales = {
"total_revenue": 847_320.00,
"total_units": 14_892,
"by_region": {
"Midwest": {"revenue": 312_400.00, "units": 5_821},
"Northeast": {"revenue": 287_600.00, "units": 4_903},
"South": {"revenue": 164_800.00, "units": 2_988},
"Canada": {"revenue": 82_520.00, "units": 1_180},
},
"by_category": {
"Commercial Paper": {"revenue": 243_100.00, "units": 6_204},
"Toner & Ink": {"revenue": 198_400.00, "units": 2_891},
"Desk Supplies": {"revenue": 152_200.00, "units": 3_112},
"Filing & Storage": {"revenue": 133_900.00, "units": 1_947},
"Technology": {"revenue": 119_720.00, "units": 738},
},
}
# Generate the report
generate_market_intelligence_report(
news_articles=news_articles,
exchange_data=exchange_data,
competitor_data=competitor_data,
acme_sales=acme_sales,
output_path=output_path,
)
# Summary
print(f"\nMarket intelligence pull complete.")
print(f" Industry articles: {len(news_articles)}")
print(f" USD/CAD rate: {exchange_data.get('usd_to_cad', 'N/A'):.4f}")
if competitor_data:
print(f" Competitor: {competitor_data.get('company_name', 'N/A')}")
print(f" Report: {output_path}")
if __name__ == "__main__":
main()
The Result
The first time Priya runs the full pipeline, it pulls seven relevant news articles (two about supply chain tightness in commercial paper, one about a major distributor's earnings miss, four about industry pricing trends), retrieves the USD/CAD rate, and formats Acme's internal numbers into a single coherent brief.
She emails it to Sandra on Friday afternoon. Sandra replies within the hour.
"This is exactly what I needed. Can you set this up to run automatically every Friday morning?"
That question leads directly to Chapter 22.
What Priya Learned
API calls are just function calls with a network round trip. Once she got past the authentication setup and understood the response structure, the actual code was surprisingly compact.
Error handling is non-negotiable. The first version of her news function crashed when NewsAPI returned an articles: null on a weekend rate limit. The second version checked for it. The third version would have been the same — until she added the try/except block around each individual query, which allowed the rest of the pipeline to continue even when one query failed.
Combining data sources creates value. Any one of these data streams — news, exchange rates, competitor financials, internal sales — was limited on its own. Together, they told a story that internal data alone never could.
API keys need a home. Before this project, Priya was copy-pasting API keys into scripts when she needed them and deleting them after. The .env file approach is more organized, more secure, and easier to maintain across multiple scripts.
Discussion Questions
-
Priya's news queries are fairly broad. What are the tradeoffs between a narrow query (very specific, fewer false positives, might miss relevant news) and a broad query (more results, more noise)?
-
The competitor financial data comes from Alpha Vantage's free tier, which has a 25-requests-per-day limit. If Priya's pipeline needs to monitor 10 competitor tickers daily, what are her options?
-
Priya uses static internal sales data in the demo. How would you modify
main()to read the sales data from a real CSV file that gets updated weekly? -
Sandra asked for this to run automatically every Friday morning. Before looking at Chapter 22, sketch out what you think that would involve.