Financial Intelligence Meta-Layer (FIML)¶

class DataArbitrationEngine:  
    """  
    Intelligently routes data requests to optimal providers  
    """  

    async def arbitrate_request(  
        self,  
        asset: Asset,  
        data_type: DataType,  
        requirements: Requirements  
    ) -> ArbitrationPlan:  
        """  
        Returns an execution plan specifying:  
        - Primary provider  
        - Fallback providers (ordered)  
        - Merge strategy if multiple sources needed  
        - Expected latency and freshness  
        """  

        # Step 1: Get provider health scores  
        provider_scores = await self._score_providers(  
            asset, data_type, requirements  
        )  

        # Step 2: Check freshness requirements  
        freshness_filtered = self._filter_by_freshness(  
            provider_scores, requirements.max_staleness  
        )  

        # Step 3: Check regional latency  
        latency_optimized = self._optimize_by_latency(  
            freshness_filtered, requirements.user_region  
        )  

        # Step 4: Check rate limits  
        available_providers = await self._check_rate_limits(  
            latency_optimized  
        )  

        # Step 5: Create execution plan  
        return ArbitrationPlan(  
            primary=available_providers[0],  
            fallbacks=available_providers[1:3],  
            merge_strategy=self._get_merge_strategy(data_type),  
            estimated_latency_ms=available_providers[0].latency_p95  
        )  

def calculate_provider_score(  
    provider: Provider,  
    asset: Asset,  
    data_type: DataType,  
    region: Region  
) -> ProviderScore:  
    """  
    Real-time provider scoring  
    """  

    # Freshness score  
    last_update = provider.get_last_update(asset, data_type)  
    age_minutes = (now() - last_update).total_seconds() / 60  
    freshness_score = max(0, 100 * (1 - age_minutes / 60))  

    # Latency score (regional)  
    latency_p95 = provider.get_latency_p95(region)  
    latency_score = max(0, 100 * (1 - latency_p95 / 5000))  

    # Uptime score  
    uptime_score = provider.get_uptime_24h() * 100  

    # Completeness score  
    completeness_score = provider.get_completeness(data_type) * 100  

    # Historical reliability  
    reliability_score = provider.get_success_rate() * 100  

    # Weighted total  
    total_score = (  
        freshness_score * 0.30 +  
        latency_score * 0.25 +  
        uptime_score * 0.20 +  
        completeness_score * 0.15 +  
        reliability_score * 0.10  
    )  

    return ProviderScore(  
        total=total_score,  
        freshness=freshness_score,  
        latency=latency_score,  
        uptime=uptime_score,  
        completeness=completeness_score,  
        reliability=reliability_score  
    )  

async def execute_with_fallback(  
    plan: ArbitrationPlan,  
    request: DataRequest  
) -> DataResponse:  
    """  
    Execute with automatic fallback  
    """  

    providers = [plan.primary] + plan.fallbacks  

    for provider in providers:  
        try:  
            response = await provider.fetch(  
                request,   
                timeout=plan.timeout_ms  
            )  

            # Validate response  
            if response.is_valid() and response.is_fresh():  
                return response  

        except (TimeoutError, RateLimitError, ProviderError) as e:  
            logger.warning(  
                f"Provider {provider.name} failed: {e}. "  
                f"Falling back to next provider."  
            )  
            await self._record_failure(provider, request, e)  
            continue  

    # All providers failed  
    raise NoProviderAvailableError(  
        f"All providers failed for {request.asset}"  
    )  

class MergeStrategy:  
    """  
    Intelligently merge data from multiple providers  
    """  

    async def merge_multi_provider(  
        self,  
        responses: List[ProviderResponse]  
    ) -> MergedResponse:  
        """  
        Merge strategy examples:  
        - OHLCV: Average price, sum volume  
        - Fundamentals: Take most recent filing  
        - Sentiment: Weighted average by source credibility  
        """  

        if not responses:  
            raise ValueError("No responses to merge")  

        # Determine merge strategy by data type  
        data_type = responses[0].data_type  

        if data_type == DataType.OHLCV:  
            return self._merge_ohlcv(responses)  
        elif data_type == DataType.FUNDAMENTALS:  
            return self._merge_fundamentals(responses)  
        elif data_type == DataType.SENTIMENT:  
            return self._merge_sentiment(responses)  
        else:  
            # Default: take most recent  
            return max(responses, key=lambda r: r.timestamp)  

    def _merge_ohlcv(  
        self,   
        responses: List[ProviderResponse]  
    ) -> MergedResponse:  
        """  
        OHLCV merge logic:  
        - Open: Take earliest timestamp  
        - High: Max of all highs  
        - Low: Min of all lows  
        - Close: Take latest timestamp  
        - Volume: Sum all volumes (deduplicate exchanges)  
        """  

        return MergedResponse(  
            open=min(responses, key=lambda r: r.timestamp).open,  
            high=max(r.high for r in responses),  
            low=min(r.low for r in responses),  
            close=max(responses, key=lambda r: r.timestamp).close,  
            volume=sum(r.volume for r in responses),  
            source_count=len(responses),  
            sources=[r.provider for r in responses],  
            confidence=self._calculate_confidence(responses)  
        )  

class ConflictResolver:  
    """  
    Resolve conflicts when providers give different answers  
    """  

    async def resolve_conflict(  
        self,  
        asset: Asset,  
        field: str,  
        values: List[ProviderValue]  
    ) -> ResolvedValue:  
        """  
        Resolution strategies:  
        1. Statistical outlier detection  
        2. Provider trust scores  
        3. Historical accuracy  
        4. Timestamp-based recency  
        """  

        # Calculate statistical outliers  
        median = np.median([v.value for v in values])  
        std = np.std([v.value for v in values])  

        # Filter outliers (> 2 std from median)  
        filtered = [  
            v for v in values   
            if abs(v.value - median) <= 2 * std  
        ]  

        # Weight by provider trust score  
        weighted_sum = sum(  
            v.value * v.provider.trust_score   
            for v in filtered  
        )  
        weighted_count = sum(  
            v.provider.trust_score   
            for v in filtered  
        )  

        final_value = weighted_sum / weighted_count  

        # Calculate confidence based on agreement  
        variance = np.var([v.value for v in filtered])  
        confidence = 1.0 / (1.0 + variance)  

        return ResolvedValue(  
            value=final_value,  
            confidence=confidence,  
            source_count=len(filtered),  
            discarded_outliers=len(values) - len(filtered),  
            method="weighted_average_with_outlier_removal",  
            providers=[v.provider.name for v in filtered]  
        )  

from lark import Lark, Transformer, v_args  

FK_DSL_GRAMMAR = r"""  
    ?start: statement+  

    statement: evaluate_stmt  
             | compare_stmt  
             | macro_stmt  
             | correlate_stmt  
             | scan_stmt  

    evaluate_stmt: "EVALUATE" asset ":" metric_list  

    compare_stmt: "COMPARE" asset ("vs" asset)+ "ON:" metric_list  

    macro_stmt: "MACRO:" indicator_list "→" analysis_type "ON" asset  

    correlate_stmt: "CORRELATE" asset "WITH" asset_list ("WINDOW" timeframe)?  

    scan_stmt: "SCAN" market "WHERE" condition_list  

    asset: SYMBOL | COIN  

    metric_list: metric ("," metric)*  

    metric: simple_metric  
          | metric_with_params  

    simple_metric: "PRICE" | "VOLUME" | "MARKETCAP" | "PE" | "BETA"  
                 | "RSI" | "MACD" | "MOMENTUM" | "LIQUIDITY" | "SENTIMENT"  

    metric_with_params: METRIC_NAME "(" params ")"  

    params: param ("," param)*  

    param: NUMBER | timeframe | STRING  

    timeframe: NUMBER TIMEUNIT  

    TIMEUNIT: "d" | "h" | "w" | "m" | "y"  

    indicator_list: indicator ("," indicator)*  

    indicator: "US10Y" | "CPI" | "VIX" | "DXY" | "UNEMPLOYMENT"  
             | "GDP" | "FED_RATE"  

    analysis_type: "REGRESSION" | "CORRELATION" | "CAUSALITY" | "IMPACT"  

    condition_list: condition ("AND" condition)*  

    condition: metric comparison_op value  

    comparison_op: ">" | "<" | ">=" | "<=" | "=" | "!="  

    market: "NYSE" | "NASDAQ" | "LSE" | "TSE" | "CRYPTO"  

    asset_list: asset ("," asset)*  

    value: NUMBER | STRING  

    SYMBOL: /[A-Z][A-Z0-9]{0,5}/  
    COIN: /[A-Z]{2,5}/  
    METRIC_NAME: /[A-Z_]+/  
    NUMBER: /[0-9]+(\.[0-9]+)?/  
    STRING: /"[^"]*"/  

    %import common.WS  
    %ignore WS  
"""  

class FKDSLTransformer(Transformer):  
    @v_args(inline=True)  
    def evaluate_stmt(self, asset, metrics):  
        return {  
            "type": "evaluate",  
            "asset": asset,  
            "metrics": metrics  
        }  

    @v_args(inline=True)  
    def compare_stmt(self, *args):  
        assets = [a for a in args if isinstance(a, dict) and "symbol" in a]  
        metrics = args[-1]  
        return {  
            "type": "compare",  
            "assets": assets,  
            "metrics": metrics  
        }  

    @v_args(inline=True)  
    def macro_stmt(self, indicators, analysis_type, asset):  
        return {  
            "type": "macro",  
            "indicators": indicators,  
            "analysis": analysis_type,  
            "target": asset  
        }  

    # ... more transformers  

fkdsl_parser = Lark(FK_DSL_GRAMMAR, parser='lalr', transformer=FKDSLTransformer())  

def parse_fkdsl(query: str) -> Dict:  
    """Parse FK-DSL query into execution plan"""  
    try:  
        return fkdsl_parser.parse(query)  
    except Exception as e:  
        raise FKDSLParseError(f"Invalid FK-DSL query: {e}")  

class FKDSLExecutor:  
    """  
    Execute parsed FK-DSL queries  
    """  

    async def execute(self, parsed_query: Dict) -> TaskID:  
        """  
        Execute FK-DSL query and return task ID  
        """  

        # Create execution plan  
        plan = self._create_execution_plan(parsed_query)  

        # Create task  
        task = await self.task_manager.create_task(  
            type="fkdsl_execution",  
            plan=plan  
        )  

        # Schedule execution  
        await self.scheduler.schedule_task(task.id, plan)  

        return task.id  

    def _create_execution_plan(self, parsed_query: Dict) -> ExecutionPlan:  
        """  
        Create DAG of execution steps  
        """  

        query_type = parsed_query["type"]  

        if query_type == "evaluate":  
            return self._plan_evaluate(parsed_query)  
        elif query_type == "compare":  
            return self._plan_compare(parsed_query)  
        elif query_type == "macro":  
            return self._plan_macro(parsed_query)  
        # ... more planners  

    def _plan_evaluate(self, query: Dict) -> ExecutionPlan:  
        """  
        Plan evaluation query execution  

        DAG structure:  
        1. Fetch cached data (parallel)  
        2. Compute metrics (parallel, depends on #1)  
        3. Generate narrative (depends on #2)  
        """  

        asset = query["asset"]  
        metrics = query["metrics"]  

        # Step 1: Fetch data  
        fetch_steps = []  
        for metric in metrics:  
            if metric in ["PRICE", "VOLUME", "MARKETCAP"]:  
                fetch_steps.append(  
                    Step(  
                        id=f"fetch_{metric}",  
                        type="fetch_cached",  
                        params={"asset": asset, "field": metric},  
                        dependencies=[]  
                    )  
                )  

        # Step 2: Compute metrics  
        compute_steps = []  
        for metric in metrics:  
            if metric.startswith("VOLATILITY"):  
                compute_steps.append(  
                    Step(  
                        id="compute_volatility",  
                        type="compute_volatility",  
                        params={"asset": asset, "period": metric.params[0]},  
                        dependencies=["fetch_PRICE"]  
                    )  
                )  
            # ... more metric computations  

        # Step 3: Generate narrative  
        narrative_step = Step(  
            id="generate_narrative",  
            type="narrative_generation",  
            params={"query": query},  
            dependencies=[s.id for s in fetch_steps + compute_steps]  
        )  

        return ExecutionPlan(  
            steps=fetch_steps + compute_steps + [narrative_step],  
            estimated_duration_ms=2000  
        )  

@ray.remote  
class FundamentalsWorker:  
    """  
    Specializes in fundamental financial data  
    """  

    def __init__(self):  
        self.providers = [  
            AlphaVantageProvider(),  
            FMPProvider(),  
            YFinanceProvider()  
        ]  
        self.cache = FundamentalsCache()  

    async def analyze(  
        self,   
        asset: Asset,   
        depth: str = "standard"  
    ) -> FundamentalsAnalysis:  
        """  
        Fetch and analyze fundamental data  
        """  

        # Fetch from multiple providers  
        results = await asyncio.gather(*[  
            provider.get_fundamentals(asset)  
            for provider in self.providers  
        ], return_exceptions=True)  

        # Merge results  
        merged = self._merge_fundamentals(results)  

        # Compute derived metrics  
        ratios = self._compute_ratios(merged)  

        # Quality score  
        quality = self._assess_quality(merged, results)  

        return FundamentalsAnalysis(  
            data=merged,  
            ratios=ratios,  
            quality=quality,  
            sources=[p.name for p in self.providers],  
            timestamp=now()  
        )  

    def _compute_ratios(self, data: FundamentalsData) -> FinancialRatios:  
        """  
        Compute financial ratios from raw data  
        """  
        return FinancialRatios(  
            pe=data.market_cap / data.earnings,  
            pb=data.market_cap / data.book_value,  
            ps=data.market_cap / data.revenue,  
            debt_to_equity=data.total_debt / data.equity,  
            current_ratio=data.current_assets / data.current_liabilities,  
            roe=data.net_income / data.equity,  
            roa=data.net_income / data.total_assets,  
            # ... more ratios  
        )  

@ray.remote  
class TechnicalWorker:  
    """  
    Specializes in technical analysis  
    """  

    def __init__(self):  
        self.indicators = {  
            "RSI": RSIIndicator(),  
            "MACD": MACDIndicator(),  
            "BOLLINGER": BollingerBandsIndicator(),  
            "EMA": EMAIndicator(),  
            "SMA": SMAIndicator(),  
            "STOCHASTIC": StochasticIndicator(),  
            "ATR": ATRIndicator(),  
            "OBV": OBVIndicator(),  
        }  

    async def analyze(  
        self,  
        asset: Asset,  
        indicators: List[str],  
        period: int = 14  
    ) -> TechnicalAnalysis:  
        """  
        Compute technical indicators  
        """  

        # Fetch OHLCV data  
        ohlcv = await self._fetch_ohlcv(asset, lookback=100)  

        # Compute requested indicators  
        results = {}  
        for indicator_name in indicators:  
            indicator = self.indicators.get(indicator_name)  
            if indicator:  
                results[indicator_name] = indicator.compute(  
                    ohlcv,   
                    period=period  
                )  

        # Generate signals  
        signals = self._generate_signals(results)  

        return TechnicalAnalysis(  
            indicators=results,  
            signals=signals,  
            timestamp=now()  
        )  

    def _generate_signals(  
        self,   
        indicators: Dict[str, IndicatorResult]  
    ) -> List[Signal]:  
        """  
        Generate trading signals from indicators  
        (For informational purposes only, NOT advice)  
        """  
        signals = []  

        # RSI signal  
        if "RSI" in indicators:  
            rsi_value = indicators["RSI"].value  
            if rsi_value > 70:  
                signals.append(Signal(  
                    type="RSI_OVERBOUGHT",  
                    strength="high",  
                    description="RSI above 70 indicates overbought conditions"  
                ))  
            elif rsi_value < 30:  
                signals.append(Signal(  
                    type="RSI_OVERSOLD",  
                    strength="high",  
                    description="RSI below 30 indicates oversold conditions"  
                ))  

        # MACD signal  
        if "MACD" in indicators:  
            macd = indicators["MACD"]  
            if macd.macd > macd.signal and macd.histogram > 0:  
                signals.append(Signal(  
                    type="MACD_BULLISH_CROSSOVER",  
                    strength="medium",  
                    description="MACD crossed above signal line"  
                ))  

        # ... more signal logic  

        return signals  

@ray.remote  
class MacroWorker:  
    """  
    Specializes in macroeconomic analysis  
    """  

    def __init__(self):  
        self.indicators = {  
            "US10Y": TreasuryYieldIndicator(),  
            "CPI": CPIIndicator(),  
            "VIX": VIXIndicator(),  
            "DXY": DollarIndexIndicator(),  
            "FED_RATE": FedFundsRateIndicator(),  
            "UNEMPLOYMENT": UnemploymentIndicator(),  
        }  

    async def analyze(  
        self,  
        target_asset: Asset,  
        indicators: List[str],  
        window: str = "90d"  
    ) -> MacroAnalysis:  
        """  
        Analyze macro impact on target asset  
        """  

        # Fetch macro data  
        macro_data = await asyncio.gather(*[  
            self.indicators[ind].fetch(window=window)  
            for ind in indicators  
        ])  

        # Fetch target asset data  
        target_data = await self._fetch_target_data(target_asset, window)  

        # Run correlation analysis  
        correlations = self._compute_correlations(  
            macro_data,   
            target_data  
        )  

        # Run regression  
        regression = self._run_regression(macro_data, target_data)  

        # Generate macro context narrative  
        narrative = self._generate_macro_narrative(  
            correlations,   
            regression,  
            current_macro_data=await self._fetch_current_macro()  
        )  

        return MacroAnalysis(  
            correlations=correlations,  
            regression=regression,  
            narrative=narrative,  
            timestamp=now()  
        )  

@ray.remote  
class SentimentWorker:  
    """  
    Specializes in sentiment analysis  
    """  

    def __init__(self):  
        self.news_sources = [  
            AlphaVantageNewsProvider(),  
            FinancialModelingPrepNewsProvider(),  
            CustomNewsScraperProvider()  
        ]  
        self.nlp_model = load_model("finbert-sentiment")  

    async def analyze(  
        self,  
        asset: Asset,  
        lookback: str = "7d"  
    ) -> SentimentAnalysis:  
        """  
        Analyze news and social sentiment  
        """  

        # Fetch news  
        news_articles = await self._fetch_news(asset, lookback)  

        # Run NLP sentiment analysis  
        sentiments = []  
        for article in news_articles:  
            sentiment_score = self.nlp_model.predict(article.text)  
            sentiments.append({  
                "title": article.title,  
                "score": sentiment_score,  
                "source": article.source,  
                "timestamp": article.timestamp  
            })  

        # Aggregate sentiment  
        aggregated = self._aggregate_sentiment(sentiments)  

        # Detect sentiment shifts  
        shifts = self._detect_sentiment_shifts(sentiments)  

        return SentimentAnalysis(  
            aggregated=aggregated,  
            articles=sentiments[:10],  # top 10  
            shifts=shifts,  
            timestamp=now()  
        )  

    def _aggregate_sentiment(  
        self,   
        sentiments: List[Dict]  
    ) -> AggregatedSentiment:  
        """  
        Aggregate individual sentiment scores  
        """  
        scores = [s["score"] for s in sentiments]  

        return AggregatedSentiment(  
            mean=np.mean(scores),  
            median=np.median(scores),  
            std=np.std(scores),  
            positive_ratio=sum(1 for s in scores if s > 0.5) / len(scores),  
            negative_ratio=sum(1 for s in scores if s < -0.5) / len(scores),  
            neutral_ratio=sum(1 for s in scores if -0.5 <= s <= 0.5) / len(scores),  
            count=len(scores)  
        )