Pricing Power Analysis

import os
from logging import Logger, getLogger
from typing import Dict, List, Optional

import matplotlib.gridspec as gridspec
import matplotlib.pyplot as plt
import matplotlib.colors as mcolors
from matplotlib.colors import LinearSegmentedColormap
import numpy as np
import pandas as pd
from pandas import DataFrame, merge
import seaborn as sns

from src.tool import plot_top_companies_by_sector, identify_basket_and_plot_confidence, analyze_basket_weekly_with_labels

from bigdata_client import Bigdata
from bigdata_client.models.entities import Company
from bigdata_client.models.search import DocumentType

from bigdata_research_tools.labeler.screener_labeler import ScreenerLabeler
from bigdata_research_tools.search.screener_search import search_by_companies
from bigdata_research_tools.excel import ExcelManager

# Setup for Plotly in Colab
import plotly.offline as pyo
from plotly.subplots import make_subplots
import plotly
import plotly.io as pio
import plotly.graph_objects as go

# Enable offline mode for Colab
pyo.init_notebook_mode(connected=True)

# Set renderer
pio.renderers.default = 'colab'


# Define output file paths for our results
output_dir = "output"
os.makedirs(output_dir, exist_ok=True)

export_path = f"{output_dir}/pricing_power_analysis_results.xlsx"

# ===== Fixed Parameters =====

# Pricing Power Theme
pricing_power_theme = "Examples of pricing power for a company"

# Lack of Pricing Power Theme
no_pricing_power_theme = "Examples of lack of pricing power for a company"

# Sentences expressing positive pricing power narratives
pricing_power_sentences_list = [
    'Company has significant pricing power',
    'Firm can set higher prices without losing customers',
    'Organization can set prices above competitors',
    'Corporation can increase prices without affecting sales',
    'Firm can implement price hikes without losing clientele',
    'Business has the ability to control prices',
    'Enterprise can leverage pricing power',
    'Enterprise can sustain higher prices',
    'Business can charge premium prices',
    'Corporation can maintain high prices without losing customers',
    'Business can increase prices without affecting sales',
    'Company can elevate prices without a drop in demand',
    'Enterprise can adjust prices without losing market share',
    'Company can implement price hikes successfully',
    'Firm can raise prices without reducing demand',
    'Corporation can set prices independently',
    'Company can dictate prices in the market',
    'Company can manage prices without losing clients',
    'Organization can increase costs without reducing sales volume',
    'Organization can increase prices without customer backlash',
    'Corporation has the ability to set higher prices',
    'Business can set premium prices without losing customers',
    'Firm has the power to influence prices',
    'Business can control pricing strategy effectively',
    'Enterprise can adjust prices upwards without losing market share',
    'Organization can raise prices without demand impact',
    'Firm can adjust pricing without negative impact'
]

# Sentences expressing lack of pricing power narratives
no_pricing_power_sentences_list = [
    'Company has no pricing power',
    'Firm cannot set higher prices without losing customers',
    'Organization cannot set prices above competitors',
    'Corporation cannot increase prices without affecting sales',
    'Firm cannot implement price hikes without losing clientele',
    'Business lacks the ability to control prices',
    'Enterprise cannot leverage pricing power',
    'Enterprise cannot sustain higher prices',
    'Business cannot charge premium prices',
    'Corporation cannot maintain high prices without losing customers',
    'Business cannot increase prices without affecting sales',
    'Company cannot elevate prices without a drop in demand',
    'Enterprise cannot adjust prices without losing market share',
    'Company cannot implement price hikes successfully',
    'Firm cannot raise prices without reducing demand',
    'Corporation cannot set prices independently',
    'Company cannot dictate prices in the market',
    'Company cannot manage prices without losing clients',
    'Organization cannot increase costs without reducing sales volume',
    'Organization cannot increase prices without customer backlash',
    'Corporation has no ability to set higher prices',
    'Business cannot set premium prices without losing customers',
    'Firm has no power to influence prices',
    'Business cannot control pricing strategy effectively',
    'Enterprise cannot adjust prices upwards without losing market share',
    'Organization cannot raise prices without demand impact',
    'Firm cannot adjust pricing without negative impact'
]

# Labels to categorize pricing power related sentences
pricing_power_labels_list = [
    "Successful Price Increase",
    "Strong Brand Recognition",
    "Launch of Differentiated Product",
    "Customer Loyalty Confirmed",
    "Effective Cost Pass-Through",
    "Competitor Price Follow",
    "High Demand Despite Price Hike",
    "Monopoly or Legal Advantage Maintained",
    "Premium Market Position Reinforced",
    "Switching Costs Recognized by Market"
]

# Labels to categorize lack of pricing power related sentences
no_pricing_power_labels_list = [
    "Failed Price Increase",
    "Weak Brand Recognition",
    "Launch of Commoditized Product",
    "Customer Loyalty Lost",
    "Ineffective Cost Pass-Through",
    "Competitor Price Undercut",
    "Demand Drop After Price Hike",
    "Loss of Market Position",
    "Commodity Market Position",
    "Low Switching Costs",
    "Lack of Pricing Power",
    "Weak Pricing Control",
    "Limited Pricing Influence",
    "Poor Pricing Strategy",
    "No Pricing Authority"
]

# Document Configuration
document_type = DocumentType.NEWS

# ===== Customizable Parameters =====

# Company Universe (from Watchlist)
# We get Top US 100 watchlist from Bigdata.com
my_watchlist_id = "44118802-9104-4265-b97a-2e6d88d74893"
watchlist = bigdata.watchlists.get(my_watchlist_id)
companies = bigdata.knowledge_graph.get_entities(watchlist.items)

# Choose MT Newswires as a news source
sources = bigdata.knowledge_graph.find_sources("MT Newswires")
tech_news_ids = [source.id for source in sources if "MT Newswires" == source.name]

# LLM Specification
llm_model = "openai::gpt-4o-mini"

# Search Frequency
search_frequency='W'

# Enable/Disable Reranker
rerank_threshold = None

# Specify Time Range
start_date = "2024-01-01"
end_date = "2024-06-01"

# Document Limits
document_limit = 10

# Others
batch_size = 10

df_sentences_positive = search_by_companies(
    companies=companies,
    sentences=pricing_power_sentences_list,
    start_date=start_date,
    end_date=end_date,
    scope=document_type,
    sources=tech_news_ids,
    rerank_threshold=rerank_threshold,
    freq=search_frequency,
    document_limit=document_limit,
    batch_size=batch_size,
)

df_sentences_negative = search_by_companies(
    companies=companies,
    sentences=no_pricing_power_sentences_list,
    start_date=start_date,
    end_date=end_date,
    scope=document_type,
    sources=tech_news_ids,
    rerank_threshold=rerank_threshold,
    freq=search_frequency,
    document_limit=document_limit,
    batch_size=batch_size,
)

labeler = ScreenerLabeler(llm_model=llm_model)

df_labels_positive = labeler.get_labels(
    main_theme=pricing_power_theme,
    labels=pricing_power_labels_list,
    texts=df_sentences_positive["masked_text"].tolist(),
)

# Merge and process results for positive pricing power
df_positive = merge(df_sentences_positive, df_labels_positive, left_index=True, right_index=True)
df_positive_relevant = labeler.post_process_dataframe(df_positive)

df_labels_negative = labeler.get_labels(
    main_theme=no_pricing_power_theme,
    labels=no_pricing_power_labels_list,
    texts=df_sentences_negative["masked_text"].tolist(),
)

# Merge and process results for negative pricing power
df_negative = merge(df_sentences_negative, df_labels_negative, left_index=True, right_index=True)
df_negative_relevant = labeler.post_process_dataframe(df_negative)

plot_top_companies_by_sector(df_positive_relevant, min_companies=1, title_suffix="(Pricing Power)")

plot_top_companies_by_sector(df_negative_relevant, min_companies=1, title_suffix="(Lack of Pricing Power)")

companies_basket = identify_basket_and_plot_confidence(df_positive_relevant, df_negative_relevant, basket_size=30, sector_column='Sector')

weekly_exposure = analyze_basket_weekly_with_labels(df_positive_relevant, df_negative_relevant, companies_basket, start_date, end_date, 'Sector')

try:
    # Create the Excel manager
    excel_manager = ExcelManager()

    # Define the dataframes and their sheet configurations
    df_args = [
        (df_positive_relevant, "Positive Pricing Power", (0, 0)),
        (df_negative_relevant, "Negative Pricing Power", (0, 0)),
        (companies_basket, "Pricing Power Basket", (2, 4)),
        (weekly_exposure, "Weekly Exposure Analysis", (2, 2))
    ]

    # Save the workbook
    excel_manager.save_workbook(df_args, export_path)

except Exception as e:
    print(f"Warning while exporting to excel: {e}")