Quiver Quantitative

from AlgorithmImports import *


class QuiverCongressDataAlgorithm(QCAlgorithm):

    def initialize(self) -> None:
        self.set_start_date(2024, 9, 1)
        self.set_end_date(2024, 12, 31)
        self.set_cash(100000)

        # Requesting data per underlying data to subscribe to updated congress members' trade information
        aapl = self.add_equity("AAPL", Resolution.DAILY).symbol
        quiver_congress_symbol = self.add_data(QuiverCongress, aapl).symbol

        # Historical data
        history = self.history(QuiverCongress, quiver_congress_symbol, 60, Resolution.DAILY)
        self.debug(f"We got {len(history)} items from our history request");

    def on_data(self, slice: Slice) -> None:
        congress_by_symbol = slice.Get(QuiverCongress)

        # Determine net direction of Congress trades for each security to estimate the sentiment direction due to political factor
        net_quantity_by_symbol = {}
        for symbol, points in congress_by_symbol.items():
            symbol = symbol.underlying
            if symbol not in net_quantity_by_symbol:
                net_quantity_by_symbol[symbol] = 0
            # Voting weight is by order size
            for point in points:
                net_quantity_by_symbol[symbol] += (1 if point.transaction == OrderDirection.BUY else -1) * point.amount
            
        for symbol, net_quantity in net_quantity_by_symbol.items():
            if net_quantity == 0:
                self.liquidate(symbol)
                continue
            # Buy when Congress members have bought, short otherwise
            self.set_holdings(symbol, 1 if net_quantity > 0 else -1)

public class QuiverCongressDataAlgorithm : QCAlgorithm
{
    public override void Initialize()
    {
        SetStartDate(2024, 9, 1);
        SetEndDate(2024, 12, 31);
        SetCash(100000);
        
        // Requesting data per underlying data to subscribe to updated congress members' trade information
        var aapl = AddEquity("AAPL", Resolution.Daily).Symbol;
        var quiverCongressSymbol = AddData<QuiverCongress>(aapl).Symbol;

        // Historical data
        var history = History<QuiverCongress>(quiverCongressSymbol, 60, Resolution.Daily);
        Debug($"We got {history.Count()} items from our history request");
    }

    
    public override void OnData(Slice slice)
    {
        var congressBySymbol = slice.Get<QuiverCongress>();
        
        // Determine net direction of Congress trades for each security to estimate the sentiment direction due to political factor
        var netQuantityBySymbol = new Dictionary<Symbol, decimal>();
        foreach (var (s, points) in congressBySymbol)
        {
            var symbol = s.Underlying;
            if (!netQuantityBySymbol.ContainsKey(symbol))
            {
                netQuantityBySymbol[symbol] = 0m;
            }
            // Voting weight is by order size
            foreach(QuiverCongressDataPoint point in points)
            {
                netQuantityBySymbol[symbol] += (point.Transaction == OrderDirection.Buy ? 1 : -1) * (point.Amount ?? 0m);
            }
        }
        
        foreach (var (symbol, netQuantity) in netQuantityBySymbol)
        {
            if (netQuantity == 0)
            {
                Liquidate(symbol);
                continue;
            }
            // Buy when Congress members have bought, short otherwise
            SetHoldings(symbol, netQuantity > 0 ? 1 : -1);
        }
    }
}

from AlgorithmImports import *


class QuiverCongressDataAlgorithm(QCAlgorithm):

    def initialize(self) -> None:
        self.set_start_date(2024, 9, 1)
        self.set_end_date(2024, 12, 31)
        self.set_cash(100000)
        self.settings.seed_initial_prices = True
        self.universe_settings.minimum_time_in_universe = timedelta(days=7)
        # Add the Quiver Congress universe.
        self.add_universe(QuiverQuantCongressUniverse, "QuiverQuantCongresssUniverse", Resolution.DAILY, self.universe_selection)
        self.add_alpha(CongressAlphaModel())
        self.set_portfolio_construction(InsightWeightingPortfolioConstructionModel(lambda time: None))
        self.add_risk_management(NullRiskManagementModel())
        self.set_execution(ImmediateExecutionModel())

    def universe_selection(self, alt_coarse: List[QuiverQuantCongressUniverse]) -> List[Symbol]:
        # Return all symbols from the Quiver Congress universe.
        return [d.symbol for d in alt_coarse]


class CongressAlphaModel(AlphaModel):
    _securities = []
    _insight_by_asset = {}

    def update(self, algorithm: QCAlgorithm, slice: Slice) -> List[Insight]:
        congress_data_by_asset = slice.get(QuiverCongress)
        if congress_data_by_asset:
            self._insight_by_asset.clear()
            net_quantity_by_asset = {}
            for congress_data_symbol, congress_data in congress_data_by_asset.items():
                security = algorithm.securities[congress_data_symbol.underlying]
                if security not in net_quantity_by_asset:
                    net_quantity_by_asset[security] = 0
                for point in congress_data:
                    net_quantity_by_asset[security] += (1 if point.transaction == OrderDirection.BUY else -1) * point.amount
            for security, net_quantity in net_quantity_by_asset.items():
                # Emit an up insight when Congress members are net buyers.
                if net_quantity > 0 and not security.holdings.is_long:
                    direction = InsightDirection.UP
                # Emit a down insight when Congress members are net sellers.
                elif net_quantity < 0 and not security.holdings.is_short:
                    direction = InsightDirection.DOWN
                else:
                    continue
                self._insight_by_asset[security] = Insight.price(security, timedelta(7), direction, weight=0.5)
        # To avoid error messages, emit the insights when the algorithm has a price for the asset.
        return [
                self._insight_by_asset.pop(security) 
                for security in list(self._insight_by_asset.keys())
                if security.price
        ]

    def on_securities_changed(self, algorithm: QCAlgorithm, changes: SecurityChanges) -> None:
        for security in changes.added_securities:
            self._securities.append(security)
            # Add the Quiver Congress subscription for the security.
            security.congress_data = algorithm.add_data(QuiverCongress, security, Resolution.DAILY)
            history = algorithm.history(QuiverCongress, security.congress_data, 14, Resolution.DAILY)
            algorithm.debug(f"We got {len(history)} items from our history request for {security}")
        for security in changes.removed_securities:
            if security in self._securities:
                if security in self._insight_by_asset:
                    self._insight_by_asset.pop(security)
                # Remove the Quiver Congress subscription for the security.
                algorithm.remove_security(security.congress_data)
                self._securities.remove(security)

public class QuiverCongressDataAlgorithm : QCAlgorithm
{
    public override void Initialize()
    {
        SetStartDate(2024, 9, 1);
        SetEndDate(2024, 12, 31);
        SetCash(100000);
        Settings.SeedInitialPrices = true;
        UniverseSettings.MinimumTimeInUniverse = TimeSpan.FromDays(7);
        // Add the Quiver Congress universe.
        AddUniverse<QuiverQuantCongressUniverse>(altCoarse => altCoarse.Select(d => d.Symbol));
        AddAlpha(new CongressAlphaModel());
        SetPortfolioConstruction(new InsightWeightingPortfolioConstructionModel(time => null));
        AddRiskManagement(new NullRiskManagementModel());
        SetExecution(new ImmediateExecutionModel());
    }
}

public class CongressAlphaModel : AlphaModel
{
    private readonly List<Security> _securities = [];
    private readonly Dictionary<Security, Insight> _insightByAsset = [];

    public override IEnumerable<Insight> Update(QCAlgorithm algorithm, Slice slice)
    {
        var congressDataByAsset = slice.Get<QuiverCongress>();
        if (congressDataByAsset.Count > 0)
        {
            _insightByAsset.Clear();
            // Aggregate the net Congress trade quantity for each security.
            var netQuantityByAsset = new Dictionary<Security, decimal>();
            foreach (var (congressDataSymbol, congressData) in congressDataByAsset)
            {
                var security = algorithm.Securities[congressDataSymbol.Underlying];
                if (!netQuantityByAsset.ContainsKey(security))
                {
                    netQuantityByAsset[security] = 0m;
                }
                // Weight each transaction by its reported amount.
                foreach (QuiverCongressDataPoint point in congressData)
                {
                    netQuantityByAsset[security] += (point.Transaction == OrderDirection.Buy ? 1 : -1) * (point.Amount ?? 0m);
                }
            }
            
            foreach (var (security, netQuantity) in netQuantityByAsset)
            {
                // Emit an up insight when Congress members are net buyers.
                InsightDirection direction;
                if (netQuantity > 0 && !security.Holdings.IsLong)
                {
                    direction = InsightDirection.Up;
                }
                // Emit a down insight when Congress members are net sellers.
                else if (netQuantity < 0 && !security.Holdings.IsShort)
                {
                    direction = InsightDirection.Down;
                }
                else
                {
                    continue;
                }
                _insightByAsset[security] = Insight.Price(security.Symbol, TimeSpan.FromDays(7), direction, weight: 0.5);
            }
        }
        // To avoid error messages, emit the insights when the algorithm has a price for the asset.
        return _insightByAsset.Keys
            .ToList()  // snapshot keys
            .Where(security => security.Price != 0m)
            .Select(security => {
                var insight = _insightByAsset[security];  // capture value
                _insightByAsset.Remove(security);  // remove
                return insight;
            })
            .ToList();
    }

    public override void OnSecuritiesChanged(QCAlgorithm algorithm, SecurityChanges changes)
    {
        foreach (var security in changes.AddedSecurities)
        {
            _securities.Add(security);
            // Add the Quiver Congress subscription for the security.
            var congressData = algorithm.AddData<QuiverCongress>(security.Symbol, Resolution.Daily);
            security.Set("CongressData", congressData);
            var history = algorithm.History<QuiverCongress>(congressData.Symbol, 14, Resolution.Daily);
            algorithm.Debug($"We got {history.Count()} items from our history request for {security}");
        }

        foreach (var security in changes.RemovedSecurities)
        {
            if (_securities.Contains(security))
            {
                _insightByAsset.Remove(security);
                // Remove the Quiver Congress subscription for the security.
                var congressData = security.Get<Security>("CongressData");
                algorithm.RemoveSecurity(congressData.Symbol);
                _securities.Remove(security);
            }
        }
    }
}