Add NUPL feature to block nodes

EBA/Blockchains/Bitcoin/BitcoinOrchestrator.cs

@@ -21,7 +21,7 @@ public BitcoinOrchestrator(
     }
 
     public async Task TraverseAsync(
-        Options options, 
+        Options options,
         CancellationToken cT)
     {
         var chainInfo = await _agent.AssertChainAsync(cT);
@@ -59,7 +59,7 @@ public async Task TraverseAsync(
 
             throw;
         }
-    }
+    }    
 
     public async Task DeDupAsync(
         Options options,

EBA/Blockchains/Bitcoin/ChainModel/BlockMetadata.cs

@@ -90,4 +90,31 @@ public long ProvablyUnspendableBitcoins
 
     public long? TotalSupply { set; get; } = null;
     public long? TotalSupplyNominal { set; get; } = null;
+
+    public decimal? RealizedCap { set; get; } = null;
+    public decimal? MarketCap
+    {
+        get
+        {
+            if (TotalSupply == null || Ohlcv == null)
+                return null;
+
+            return Ohlcv.GetFiatValue((long)TotalSupply);
+        }
+    }
+
+    public decimal? NUPL
+    {
+        get
+        {
+            if (RealizedCap == null || MarketCap == null || MarketCap == 0)
+                return null;
+
+            return (MarketCap - RealizedCap) / MarketCap;
+        }
+    }
+
+    public OHLCV? Ohlcv { set; get; } = null;
+
+    // TODO: add MVRV, NVT, Thermocap
 }

EBA/Blockchains/Bitcoin/GraphModel/BlockGraph.cs

@@ -95,7 +95,9 @@ public void BuildGraph(CancellationToken ct)
         long miningReward = 0;
         foreach (var u in _coinbaseTxGraph.Outputs)
         {
-            TryGetOrAddEdge(new T2SEdge(v, new ScriptNode(u.ScriptPubKey), t, h, output: u), out var edge);
+            TryGetOrAddEdge(
+                new T2SEdge(v, new ScriptNode(u.ScriptPubKey), t, h, u),
+                out T2SEdge _);
 
             Block.ProfileCreatedOutput(u);
             miningReward += u.Value;
@@ -123,14 +125,16 @@ private void AddTxGraphToBlockGraph(TxGraph txGraph)
         {
             TryGetOrAddEdge(
                 new S2TEdge(new ScriptNode(u.PrevOut.ScriptPubKey), v, t, h, u),
-                out var edge);
+                out S2TEdge _);
 
             Block.ProfileSpentOutput(u);
         }
 
         foreach (var u in txGraph.Outputs)
         {
-            TryGetOrAddEdge(new T2SEdge(v, new ScriptNode(u.ScriptPubKey), t, h, output: u), out var edge);
+            TryGetOrAddEdge(
+                new T2SEdge(v, new ScriptNode(u.ScriptPubKey), t, h, u), 
+                out T2SEdge _);
 
             Block.ProfileCreatedOutput(u);
         }

EBA/CLI/CLI.cs

@@ -21,7 +21,8 @@ public Cli(
         Func<Options, Task> bitcoinMapMarketHandlerAsync,
         Func<Options, Task> bitcoinAddressStatsHandlerAsync,
         Func<Options, Task> bitcoinImportCypherQueriesAsync,
-        Func<Options, Task> bitcoinPostProcessGraphHandlerAsync,
+        Func<Options, Task> bitcoinPostProcessHandlerAsync,
+        Func<Options, Task> bitcoinAugmentHandlerAsync,
         Func<Options, Task> bitcoinMapSpendsHandlerAsync,
         Action<Exception, ParseResult> exceptionHandler)
     {
@@ -90,7 +91,8 @@ e is NotSupportedException ||
                 mapMarketHandlerAsync: bitcoinMapMarketHandlerAsync,
                 addressStatsHandlerAsync: bitcoinAddressStatsHandlerAsync,
                 importCypherQueriesAsync: bitcoinImportCypherQueriesAsync,
-                postProcessGraphHandlerAsync: bitcoinPostProcessGraphHandlerAsync,
+                postProcessHandlerAsync: bitcoinPostProcessHandlerAsync,
+                bitcoinAugmentHandlerAsync: bitcoinAugmentHandlerAsync,
                 mapSpendsHandlerAsync: bitcoinMapSpendsHandlerAsync)
         };
 
@@ -127,7 +129,8 @@ private Command GetBitcoinCmd(
         Func<Options, Task> mapMarketHandlerAsync,
         Func<Options, Task> addressStatsHandlerAsync,
         Func<Options, Task> importCypherQueriesAsync,
-        Func<Options, Task> postProcessGraphHandlerAsync,
+        Func<Options, Task> postProcessHandlerAsync,
+        Func<Options, Task> bitcoinAugmentHandlerAsync,
         Func<Options, Task> mapSpendsHandlerAsync)
     {
         var cmd = new Command(
@@ -141,7 +144,8 @@ private Command GetBitcoinCmd(
             GetBitcoinMapMarketCmd(defaultOptions, mapMarketHandlerAsync),
             GetBitcoinSampleCmd(defaultOptions, sampleHandlerAsync),
             GetBitcoinAddressStatsCmd(defaultOptions, addressStatsHandlerAsync),
-            GetPostProcessGraphCmd(defaultOptions, postProcessGraphHandlerAsync),
+            GetPostProcessCmd(defaultOptions, postProcessHandlerAsync),
+            GetBitcoinAugmentCmd(defaultOptions, bitcoinAugmentHandlerAsync),
             GetBitcoinMapSpendsCmd(defaultOptions, mapSpendsHandlerAsync)
         };
         return cmd;
@@ -238,6 +242,11 @@ private Command GetBitcoinTraverseCmd(Options defaultOptions, Func<Options, Task
                 "(reference: https://neo4j.com/blog/bulk-data-import-neo4j-3-0/)"
         };
 
+        var blockMarketMappingOption = new Option<string>("--map-block-market")
+        {
+            Required = false
+        };
+
         var cmd = new Command(
             name: "traverse",
             description: "Traverses the blockchain in the defined range collecting the set metrics.")
@@ -251,7 +260,8 @@ private Command GetBitcoinTraverseCmd(Options defaultOptions, Func<Options, Task
             txoFilenameOption,
             skipGraphSerialization,
             trackTxoOption,
-            maxEntriesPerBatch
+            maxEntriesPerBatch,
+            blockMarketMappingOption
         };
 
 
@@ -301,7 +311,8 @@ private Command GetBitcoinTraverseCmd(Options defaultOptions, Func<Options, Task
                 trackTxoOption: trackTxoOption,
                 txoFilenameOption: txoFilenameOption,
                 skipGraphSerializationOption: skipGraphSerialization,
-                maxEntriesPerBatch: maxEntriesPerBatch);
+                maxEntriesPerBatch: maxEntriesPerBatch,
+                blockMarketMappingOption: blockMarketMappingOption);
 
             try
             {
@@ -445,9 +456,9 @@ private Command GetBitcoinMapMarketCmd(Options defaultOptions, Func<Options, Tas
         return cmd;
     }
 
-    private Command GetPostProcessGraphCmd(Options defaultOptions, Func<Options, Task> handlerAsync)
+    private Command GetPostProcessCmd(Options defaultOptions, Func<Options, Task> handlerAsync)
     {
-        var cmd = new Command(name: "post-process-graph");
+        var cmd = new Command(name: "post-process");
         cmd.SetAction(async (parseResult, cancellationToken) =>
         {
             var options = OptionsBinder.Build(
@@ -686,6 +697,37 @@ private Command GetBitcoinAddressStatsCmd(Options defaultOptions, Func<Options,
         return cmd;
     }
 
+    private Command GetBitcoinAugmentCmd(Options defaultOptions, Func<Options, Task> handlerAsync)
+    {
+        var ohlcvFilenameOption = new Option<string>("--ohlcv-filename")
+        {
+            Description = "A TSV file containing block metadata (height, median time) mapped to aggregated OHLCV market data."
+        };
+        var cmd = new Command(
+            name: "augment",
+            description: "Augments the graph with additional features, such as NUPL, using the output of the map-market command and other similar files.")
+        {
+            ohlcvFilenameOption
+        };
+        cmd.SetAction(async (parseResult, cancellationToken) =>
+        {
+            var options = OptionsBinder.Build(
+                parseResult,
+                workingDirOption: _workingDirOption,
+                statusFilenameOption: _statusFilenameOption,
+                augmentroOhlcvOption: ohlcvFilenameOption);
+            try
+            {
+                await handlerAsync(options);
+            }
+            catch (Exception e)
+            {
+                _exceptionHandler(e, parseResult);
+            }
+        });
+        return cmd;
+    }
+
     private Command GetBitcoinMapSpendsCmd(Options defaultOptions, Func<Options, Task> handlerAsync)
     {
         var batchesFilenameOption = new Option<string>("--batches-filename")

EBA/CLI/Config/BitcoinAugmentorOptions.cs

@@ -0,0 +1,6 @@
+namespace EBA.CLI.Config;
+
+public class BitcoinAugmentorOptions
+{
+    public string BlockOhlcvMappedFilename { init; get; }
+}

EBA/CLI/Config/BitcoinOptions.cs

@@ -9,5 +9,6 @@ public BitcoinOptions() : this(DateTimeOffset.Now.ToUnixTimeSeconds()) { }
     public BitcoinDedupOptions Dedup { init; get; } = new();
     public BitcoinGraphSampleOptions GraphSample { init; get; } = new();
     public BitcoinMapMarketOptions MapMarket { init; get; } = new();
+    public BitcoinAugmentorOptions Augmentor { init; get; } = new();
     public BitcoinMapSpendsOptions MapSpends { init; get; } = new();
 }

EBA/CLI/OptionsBinder.cs

@@ -30,6 +30,8 @@ public static Options Build(
         Option<string>? sortedScriptNodeFilenameOption = null,
         Option<string>? marketDataFilenameOption = null,
         Option<string>? outputFilenameOption = null,
+        Option<string>? blockMarketMappingOption = null,
+        Option<string>? augmentroOhlcvOption = null,
         Option<string>? batchesFilenameOption = null)
     {
         if (statusFilenameOption != null && c.GetResult(statusFilenameOption) is not null)
@@ -104,6 +106,11 @@ public static Options Build(
             BlockOhlcvMappedFilename = GetValue(defs.Bitcoin.MapMarket.BlockOhlcvMappedFilename, outputFilenameOption, c, (x) => { return Path.Join(wd, Path.GetFileName(x)); })
         };
 
+        var bitcoinAugmentorOps = new BitcoinAugmentorOptions()
+        {
+            BlockOhlcvMappedFilename = GetValue(defs.Bitcoin.Augmentor.BlockOhlcvMappedFilename, augmentroOhlcvOption, c, (x) => { return Path.Join(wd, Path.GetFileName(x)); }),
+        };
+
         var bitcoinMapSpendsOps = new BitcoinMapSpendsOptions()
         {
             BatchesFilename = GetValue(defs.Bitcoin.MapSpends.BatchesFilename, batchesFilenameOption, c, (x) => { return Path.Join(wd, Path.GetFileName(x)); })
@@ -115,6 +122,7 @@ public static Options Build(
             Dedup = bitcoinDedupOps,
             GraphSample = gsample,
             MapMarket = bitcoinMapMarketOps,
+            Augmentor = bitcoinAugmentorOps,
             MapSpends = bitcoinMapSpendsOps
         };
 

EBA/Graph/Bitcoin/BitcoinGraphAgent.cs

@@ -46,10 +46,16 @@ public async Task SerializeAsync(BitcoinGraph g, CancellationToken ct)
         await _db.SerializeAsync(g, ct);
     }
 
-    public async Task PostProcessAsync(CancellationToken ct)
+    public async Task PostBullkImportFinalizeAsync(CancellationToken ct)
     {
-        var postProcess = new PostProcess(_options, _db, _logger);
-        await postProcess.Run();
+        var finalizer = new PostBulkImportFinalizer(_options, _db, _logger);
+        await finalizer.Finalize(ct);
+    }
+
+    public async Task AddMarketData(CancellationToken ct)
+    {
+        var augmentor = new OffChain.EconomicAugmentor(_options, _db, _logger);
+        await augmentor.SetBlockMarketIndicators(ct);
     }
 
     public void Dispose()

EBA/Graph/Bitcoin/Factories/NodeFactory.cs

@@ -61,4 +61,19 @@ public static bool TryCreate(
                 $"Unexpected node type, labels: {string.Join(',', node.Labels)}");
         }
     }
+
+    public static bool TryCreate<T>(List<IRecord> records, out List<T> nodes, string nodeVar = "n") where T: Model.INode
+    {
+        nodes = [];
+        foreach (var record in records)
+        {
+            TryCreate(record[nodeVar].As<Neo4j.Driver.INode>(), out var createdNode);
+            if (createdNode is not T)
+                return false;
+
+            nodes.Add((T)createdNode);
+        }
+
+        return true;
+    }
 }

EBA/Graph/Bitcoin/OffChain/EconomicAugmentor.cs

@@ -0,0 +1,81 @@
+using EBA.Graph.Bitcoin.Factories;
+using EBA.Graph.Bitcoin.Strategies;
+using EBA.Utilities;
+
+namespace EBA.Graph.Bitcoin.OffChain;
+
+public class EconomicAugmentor(Options options, IGraphDb<BitcoinGraph> graphDb, ILogger<BitcoinGraphAgent> logger)
+{
+    private readonly Options _options = options;
+    private readonly IGraphDb<BitcoinGraph> _graphDb = graphDb;
+    private readonly ILogger<BitcoinGraphAgent> _logger = logger;
+
+    public async Task SetBlockMarketIndicators(CancellationToken ct)
+    {
+        if (!OHLCV.TryParseFile(_options.Bitcoin.Augmentor.BlockOhlcvMappedFilename, out var blockOHLCVMapping))
+        {
+            _logger.LogError(
+                "Failed to parse OHLCV data from file: {filename}",
+                _options.Bitcoin.Augmentor.BlockOhlcvMappedFilename);
+            return;
+        }
+        else
+        {
+            _logger.LogInformation(
+                "Successfully parsed OHLCV data for {count:n0} blocks from file: {filename}",
+                blockOHLCVMapping.Count,
+                _options.Bitcoin.Augmentor.BlockOhlcvMappedFilename);
+        }
+
+        var blockNodes = await GetBlockNodes(ct);
+
+        foreach (var block in blockNodes)
+            if (blockOHLCVMapping.TryGetValue(block.Key, out var ohlcv))
+                block.Value.BlockMetadata.Ohlcv = ohlcv;        
+
+        await ComputeBlockValuationMetrics(blockNodes, blockOHLCVMapping, ct);
+    }
+
+    private async Task<SortedDictionary<long, BlockNode>> GetBlockNodes(CancellationToken ct)
+    {
+        _logger.LogInformation("Fetching block nodes.");
+        var blockRecords = await _graphDb.GetNodesAsync(NodeKind.Block, ct, nodeVariable: "b");
+        _logger.LogInformation("Fetched {count:n0} block nodes.", blockRecords.Count);
+
+        var blocks = new SortedDictionary<long, BlockNode>();
+        if (NodeFactory.TryCreate<BlockNode>(blockRecords, out var blockNodes, nodeVar: "b"))
+        {
+            foreach (var blockNode in blockNodes)
+                blocks.Add(blockNode.BlockMetadata.Height, blockNode);
+        }
+        else
+        {
+            throw new Exception("Invalid node types received from database");
+        }
+
+        return blocks;
+    }
+
+    // TODO: this can be set as part of the pre-bulk import finalization step,
+    // instead of being computed and updated after the fact.
+    // This would require reordering the import steps to ensure that the OHLCV data is available before the block nodes are created.
+    private async Task ComputeBlockValuationMetrics(
+        SortedDictionary<long, BlockNode> blocks, 
+        Dictionary<long, OHLCV> blockOHLCVMapping, 
+        CancellationToken ct)
+    {
+        _logger.LogInformation("Setting realized cap for {count:n0} block nodes.", blocks.Count);
+        await _graphDb.SetRealizedCap(blocks, blockOHLCVMapping, CancellationToken.None);
+
+        _logger.LogInformation("Saving realized cap for {count:n0} block nodes.", blocks.Count);
+        var updates = blocks.Values
+            .Select(b => BlockNodeStrategy.EconomicMappings.ToDictionary(b))
+            .ToList();
+
+        await _graphDb.BulkUpdateNodePropertiesAsync(
+            NodeKind.Block,
+            nameof(BlockMetadata.Height),
+            updates,
+            ct);
+    }
+}