# ============================================================ # Project: EDGAR Research Database # Description: SEC EDGAR Research Infrastructure # # Authors: # - Manel Kammoun — manel.kammoun@uqo.ca # - Djerrycharli Tandjambianda — djerrycharli.tandjambianda@uqo.ca # - Simon-Pierre Boucher — simon-pierre.boucher@uqo.ca # # Institution: # Universite du Quebec en Outaouais (UQO) # # License: # Research / Academic Use # ============================================================ from __future__ import annotations from typing import TYPE_CHECKING import numpy as np import pyarrow as pa import pyarrow.parquet as pq from src.embeddings.models import EmbeddingModelRegistry from src.utils.config import get_config from src.utils.constants import DATA_DIRS, PARQUET_COMPRESSION, PARQUET_COMPRESSION_LEVEL from src.utils.logging import get_logger if TYPE_CHECKING: from src.storage.database import DatabaseManager logger = get_logger("factors.anchors") ANCHORS: dict[str, dict] = { "ai_adoption": { "label": "AI Internal Adoption", "category": "technology", "texts": [ ( "The company internally uses artificial intelligence, machine learning, " "generative AI, large language models, automation, predictive analytics, " "intelligent agents, and data-driven systems to improve productivity, " "operations, efficiency, decision-making, customer service, risk management, " "compliance, cybersecurity, supply chain, software development, finance, " "human resources, and other business processes." ), ( "We have implemented generative AI, large language models, and natural " "language processing capabilities across our enterprise. Our AI strategy " "includes deploying intelligent automation, computer vision, and predictive " "analytics systems to transform operations." ), ( "Investment in artificial intelligence research and development, including " "partnerships with AI technology providers, hiring of machine learning " "engineers, and deployment of AI infrastructure including GPU computing " "clusters for internal business applications." ), ], }, "ai_disruption_risk": { "label": "AI Product Disruption Risk", "category": "risk", "texts": [ ( "Artificial intelligence may reduce the scarcity, defensibility, " "differentiation, pricing power, or competitive advantage of the company's " "products or services by enabling rapid replication, automation, synthetic " "generation, low-cost substitution, or commoditization of capabilities that " "were previously difficult, expensive, or specialized." ), ( "The rapid advancement of AI technologies may render our current offerings " "less competitive. New market entrants leveraging AI could fundamentally " "alter industry dynamics, reduce barriers to entry, and erode our market " "position through AI-native products and services." ), ( "Risks associated with artificial intelligence include potential competitive " "displacement by AI-powered alternatives, obsolescence of our existing " "products, regulatory uncertainty around AI governance, algorithmic bias, " "intellectual property challenges, and emergence of AI-native competitors." ), ], }, "ai_infrastructure": { "label": "AI Infrastructure Exposure", "category": "technology", "texts": [ ( "Investment in AI computing infrastructure including data centers, GPU and " "TPU clusters, cloud computing resources, semiconductor chips, accelerator " "hardware, and high-performance computing systems dedicated to artificial " "intelligence and machine learning workloads." ), ( "Capital expenditures for AI infrastructure, including networking equipment " "for distributed AI training, specialized computing platforms for large-scale " "model development, edge computing for AI inference, and data pipeline " "infrastructure for machine learning operations." ), ( "Cloud infrastructure spending for AI workloads, including partnerships with " "hyperscale cloud providers, deployment of AI-optimized hardware, GPU supply " "chain management, and investment in next-generation computing architectures " "for artificial intelligence applications." ), ], }, "esg_commitment": { "label": "ESG Exposure", "category": "sustainability", "texts": [ ( "Environmental, social, and governance initiatives including sustainability " "reporting, corporate social responsibility programs, diversity equity and " "inclusion efforts, stakeholder engagement, and adherence to ESG frameworks " "and standards such as SASB, GRI, and TCFD." ), ( "The company is committed to ESG principles including reducing carbon " "emissions, improving energy efficiency, promoting workforce diversity, " "ensuring supply chain responsibility, ethical governance practices, and " "transparent reporting of environmental and social performance metrics." ), ( "Board-level oversight of environmental and social risk management, executive " "compensation tied to sustainability metrics, corporate governance " "improvements, and integration of ESG factors into strategic decision-making " "and capital allocation processes." ), ], }, "climate_risk": { "label": "Climate Risk", "category": "risk", "texts": [ ( "Climate change poses material risks to our operations including physical " "risks from extreme weather events, transition risks from evolving " "regulations, carbon pricing mechanisms, and shifting consumer preferences " "toward sustainable alternatives and low-carbon products." ), ( "Exposure to climate-related financial risks including greenhouse gas " "emission regulations, carbon taxes, stranded asset risk, water scarcity, " "supply chain disruption from natural disasters, and reputational risk from " "insufficient climate action and environmental degradation." ), ( "Climate risk assessment and scenario analysis aligned with TCFD " "recommendations, including evaluation of physical and transition risks " "under different warming scenarios, Scope 1, 2, and 3 greenhouse gas " "emissions measurement and reduction targets." ), ], }, "cybersecurity": { "label": "Cybersecurity Risk", "category": "risk", "texts": [ ( "Cybersecurity risks including data breaches, ransomware attacks, " "unauthorized access to systems, intellectual property theft, and " "operational disruption from cyber incidents. Investment in information " "security controls, threat detection, and incident response capabilities." ), ( "Data privacy and security compliance including GDPR, CCPA, and other data " "protection regulations. Cybersecurity governance, risk management " "frameworks, security operations centers, and employee security awareness " "training programs to protect critical infrastructure." ), ( "Critical infrastructure protection, network security, endpoint security, " "identity and access management, zero-trust architecture implementation, " "penetration testing, cyber insurance, and board-level oversight of " "cybersecurity risk management and AI-powered threat detection." ), ], }, "automation_labor": { "label": "Automation & Labor Replacement", "category": "operational", "texts": [ ( "Automation of business processes, robotic process automation, intelligent " "automation, and workforce transformation. Impact of automation on labor " "force, reskilling and upskilling programs, and the balance between human " "workers and automated systems in manufacturing and services." ), ( "Workforce restructuring driven by technological automation, including " "reduction in manual processes, deployment of automated manufacturing and " "logistics systems, and investment in human-machine collaboration " "technologies and digital transformation of work." ), ( "Labor market dynamics including talent acquisition challenges, employee " "retention, workforce planning in the context of automation, gig economy " "participation, remote work infrastructure, and the future of work " "transformation driven by AI and robotics." ), ], }, "macro_risk": { "label": "Macroeconomic Risk", "category": "risk", "texts": [ ( "Macroeconomic risks including inflation, interest rate fluctuations, " "currency exchange rate volatility, recession risk, geopolitical tensions, " "trade policy uncertainty, tariffs, and supply chain disruptions affecting " "business performance and revenue." ), ( "Economic uncertainty and market volatility, including sensitivity to GDP " "growth, consumer spending patterns, credit market conditions, commodity " "price fluctuations, and the impact of monetary and fiscal policy changes " "on business operations and profitability." ), ( "Global economic risks including sovereign debt concerns, emerging market " "instability, trade war escalation, sanctions, energy price shocks, pandemic " "impacts on economic activity, and systemic financial market risks affecting " "corporate investment and capital markets." ), ], }, } class AnchorRegistry: def __init__( self, registry: EmbeddingModelRegistry | None = None, model_name: str | None = None, ) -> None: self._registry = registry or EmbeddingModelRegistry() cfg = get_config("factors").get("factors", {}) self._model_name = model_name or cfg.get("model_name", "all-MiniLM-L6-v2") self._anchor_vectors: dict[str, np.ndarray] | None = None @property def anchor_ids(self) -> list[str]: return list(ANCHORS.keys()) def embed_anchors(self) -> dict[str, np.ndarray]: if self._anchor_vectors is not None: return self._anchor_vectors self._anchor_vectors = {} for anchor_id, info in ANCHORS.items(): texts = info["texts"] vectors = self._registry.encode( texts, model_name=self._model_name, batch_size=len(texts), show_progress=False, ) centroid = vectors.mean(axis=0) norm = np.linalg.norm(centroid) if norm > 0: centroid = centroid / norm self._anchor_vectors[anchor_id] = centroid.astype(np.float32) logger.info( "Embedded anchor '%s' (%d texts -> dim %d)", anchor_id, len(texts), centroid.shape[0], ) return self._anchor_vectors def get_anchor_matrix(self) -> tuple[list[str], np.ndarray]: vectors = self.embed_anchors() ids = list(vectors.keys()) matrix = np.stack([vectors[aid] for aid in ids]) return ids, matrix def save_to_duckdb(self, db: DatabaseManager) -> None: vectors = self.embed_anchors() db.execute("DROP TABLE IF EXISTS anchor_embeddings") db.execute(""" CREATE TABLE anchor_embeddings ( anchor_id VARCHAR PRIMARY KEY, label VARCHAR NOT NULL, category VARCHAR NOT NULL, model_name VARCHAR, dimension INTEGER, created_at TIMESTAMP DEFAULT current_timestamp ) """) for anchor_id, vec in vectors.items(): info = ANCHORS[anchor_id] db.execute( "INSERT INTO anchor_embeddings (anchor_id, label, category, model_name, dimension) " "VALUES (?, ?, ?, ?, ?)", [anchor_id, info["label"], info["category"], self._model_name, int(vec.shape[0])], ) logger.info("Saved %d anchors to DuckDB", len(vectors)) def save_to_parquet(self) -> None: vectors = self.embed_anchors() records = [] for anchor_id, vec in vectors.items(): info = ANCHORS[anchor_id] records.append({ "anchor_id": anchor_id, "label": info["label"], "category": info["category"], "model_name": self._model_name, "vector": vec.tolist(), }) table = pa.Table.from_pylist(records) out_dir = DATA_DIRS["parquet"] / "anchor_embeddings" out_dir.mkdir(parents=True, exist_ok=True) pq.write_table( table, str(out_dir / "anchor_embeddings.parquet"), compression=PARQUET_COMPRESSION, compression_level=PARQUET_COMPRESSION_LEVEL, ) logger.info("Saved anchor embeddings to parquet")