Auto tx chain detection

analytics/analytics.go

@@ -131,9 +131,9 @@ func TxFundsFlowProbability(rawData []*AllFundsFlows,
 	// Append the amounts count to the raw source outputs slice.
 	outSourceArr := appendDupsCount(rawOutSourceArr)
 
-	var totalRes []*RawResults
+	var totalRes []*rawResults
 
-	log.Debug("Calculating the transaction funds flow probabiblity...")
+	log.Debug("Calculating the transaction funds flow probability...")
 
 	allInputs := make(map[float64]int)
 	// inSourceArr contains the original list of input amounts from the tx.
@@ -150,7 +150,7 @@ func TxFundsFlowProbability(rawData []*AllFundsFlows,
 	for _, entries := range rawData {
 		for index := range entries.FundsFlow {
 			bucket := entries.FundsFlow[index]
-			g := new(RawResults)
+			g := new(rawResults)
 
 			if g.Inputs == nil {
 				g.Inputs = make(map[float64]int)
@@ -186,6 +186,12 @@ func TxFundsFlowProbability(rawData []*AllFundsFlows,
 		isMany := len(res.Inputs) > 1
 
 		for out, outSum := range res.MatchingOutputs {
+			// output amount to be used must be greater than zero. OP_RETURN
+			// scripts do not have any amounts. They hold nulldata transaction type.
+			if out <= 0 {
+				continue
+			}
+
 			if tmpRes[out] == nil {
 				tmpRes[out] = &FlowProbability{
 					uniqueInputs: make(map[float64]int),

analytics/analyticstypes.go

@@ -1,6 +1,8 @@
 package analytics
 
-import "sync"
+import (
+	"sync"
+)
 
 const (
 	// dopingElement is a placeholder value that helps guarrantee accuracy in
@@ -25,6 +27,32 @@ type Node struct {
 	Right *Node         `json:",omitempty"`
 }
 
+// Hub defines the basic unit of a transaction chain analysis graph. The Hub
+// holds details of a funds flow linked between the current TxHash and the
+// other Matched Hub(s). A chain of hubs provide the flow of funds from the current
+// output TxHash to back in time where the source of funds can be identified.
+type Hub struct {
+	// Unique details of the current output.
+	Address     string
+	Amount      float64
+	TxHash      string
+	Probability float64 `json:",omitempty"`
+
+	// setCount helps track which set whose entry has already been processed
+	// in a specific Hub.
+	setCount int
+
+	// Linked funds flow input(s).
+	Matched []Set `json:",omitempty"`
+}
+
+// Set defines a group or individual inputs that can be correctly linked to an
+// output as their source of funds.
+type Set struct {
+	Inputs          []*Hub
+	PercentOfInputs float64
+}
+
 // GroupedValues clusters together values as duplicates or other grouped values.
 // It holds the total sum and the list of the duplicates/grouped values.
 type GroupedValues struct {
@@ -47,9 +75,9 @@ type AllFundsFlows struct {
 	FundsFlow []TxFundsFlow
 }
 
-// RawResults defines some compressed solutions data needed for further processing
+// rawResults defines some compressed solutions data needed for further processing
 // of the transaction funds flow.
-type RawResults struct {
+type rawResults struct {
 	Inputs          map[float64]int
 	MatchingOutputs map[float64]*Details
 }

analytics/chains.go

@@ -3,8 +3,185 @@
 
 package analytics
 
-// type
+import (
+	"fmt"
 
-func ChainDiscovery(txHash string) {
+	"github.com/decred/dcrd/rpcclient"
+	"github.com/raedahgroup/dcrchainanalysis/v1/rpcutils"
+)
 
+// RetrieveTxData fetches a transaction data from the rpc client returns
+// processed transaction data.
+func RetrieveTxData(client *rpcclient.Client, txHash string) (*rpcutils.Transaction, error) {
+	// Return an empty Transactions object if txHash used is empty.
+	if txHash == "" {
+		return &rpcutils.Transaction{}, nil
+	}
+
+	log.Infof("Retrieving data for transaction: %s", txHash)
+
+	txData, err := rpcutils.GetTransactionVerboseByID(client, txHash)
+	if err != nil {
+		return nil, fmt.Errorf("RetrieveTxData error: failed to fetch transaction %s", txHash)
+	}
+
+	return rpcutils.ExtractRawTxTransaction(txData), nil
+}
+
+// RetrieveTxProbability returns the tx level probability values for each output.
+func RetrieveTxProbability(client *rpcclient.Client, txHash string) (
+	[]*FlowProbability, *rpcutils.Transaction, error) {
+	tx, err := RetrieveTxData(client, txHash)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	rawSolution, inputs, outputs, err := TransactionFundsFlow(tx)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return TxFundsFlowProbability(rawSolution, inputs, outputs), tx, nil
+}
+
+// ChainDiscovery returns all the possible chains associated with the tx hash used.
+func ChainDiscovery(client *rpcclient.Client, txHash string) ([]*Hub, error) {
+	tx, err := RetrieveTxData(client, txHash)
+	if err != nil {
+		return nil, err
+	}
+
+	// hubsChain defines the various paths with funds flows from a given output to
+	// back in time when the source for each path can be identified.
+	var hubsChain []*Hub
+
+	var depth = 3
+
+	for _, val := range tx.Outpoints {
+		var hubCount int
+		var stackTrace []*Hub
+		count := 1
+
+		entry := &Hub{
+			TxHash:  tx.TxID,
+			Amount:  val.Value,
+			Address: val.PkScriptData.Addresses[0],
+		}
+
+		err = handleDepths(entry, stackTrace, client, count, depth, hubCount)
+		if err != nil {
+			return nil, err
+		}
+
+		hubsChain = append(hubsChain, entry)
+	}
+
+	log.Info("Finished auto chain(s) discovery and appending all needed data")
+
+	return hubsChain, nil
+}
+
+// handleDepths recusively creates a graph-like data structure that shows the
+// funds flow path from output (UTXO) to the source of funds at the provided depth.
+func handleDepths(curHub *Hub, stack []*Hub, client *rpcclient.Client,
+	count, depth, hubCount int) error {
+	err := curHub.getDepth(client)
+	if err != nil {
+		return err
+	}
+
+	if depth == count {
+		// backtrack till we find an unprocessed Hub.
+		for {
+			count--
+			curHub = stack[len(stack)-1]
+			stack = stack[:len(stack)-1]
+
+			if hubCount+1 < len(curHub.Matched[curHub.setCount].Inputs) {
+				hubCount++
+				break
+
+			} else if curHub.setCount+1 < len(curHub.Matched) {
+				curHub.setCount++
+				hubCount = 0
+				break
+			}
+
+			if len(stack) == 0 {
+				return nil
+			}
+		}
+	}
+
+	// Adds items to the stack.
+	stack = append(stack, curHub)
+	curHub = curHub.Matched[curHub.setCount].Inputs[hubCount]
+
+	return handleDepths(curHub, stack, client, count+1, depth, hubCount)
+}
+
+// getDepth appends all the sets linked to a given output after a given amount
+// probability solution is resolved.
+func (h *Hub) getDepth(client *rpcclient.Client) error {
+	if h.TxHash == "" {
+		return nil
+	}
+
+	probability, tx, err := RetrieveTxProbability(client, h.TxHash)
+	if err != nil {
+		return err
+	}
+
+	for _, item := range probability {
+		if item.OutputAmount == h.Amount {
+			for _, entry := range item.ProbableInputs {
+				d, err := getSet(client, tx, entry)
+				if err != nil {
+					return err
+				}
+
+				h.Probability = item.LinkingProbability
+				h.Matched = append(h.Matched, d)
+			}
+		}
+	}
+	return nil
+}
+
+// The Set returned in a given output probability solution does not have a lot of
+// data, this functions reconstructs the Set adding the necessary information.
+func getSet(client *rpcclient.Client, txData *rpcutils.Transaction,
+	matchedInputs *InputSets) (set Set, err error) {
+	inputs := make([]rpcutils.TxInput, len(txData.Inpoints))
+	copy(inputs, txData.Inpoints)
+
+	for _, item := range matchedInputs.Set {
+		for i := 0; i < item.Count; i++ {
+			for k, d := range inputs {
+				if d.ValueIn == item.Amount {
+					s := &Hub{Amount: d.ValueIn, TxHash: d.TxHash}
+
+					tx, err := RetrieveTxData(client, s.TxHash)
+					if err != nil {
+						return Set{}, err
+					}
+
+					// fetch the current hub's Address.
+					for k := range tx.Outpoints {
+						if d.OutputTxIndex == tx.Outpoints[k].TxIndex {
+							s.Address = tx.Outpoints[k].PkScriptData.Addresses[0]
+						}
+					}
+
+					set.Inputs = append(set.Inputs, s)
+					set.PercentOfInputs = matchedInputs.PercentOfInputs
+
+					copy(inputs[k:], inputs[k+1:])
+					inputs = inputs[:len(inputs)-1]
+					break
+				}
+			}
+		}
+	}
+	return
 }

analytics/internals.go

@@ -25,11 +25,11 @@ func extractAmounts(data *rpcutils.Transaction) (inputs, outputs []float64) {
 	sort.Float64s(outputs)
 
 	// Add the doping element when the last entry in the slice is a duplicate.
-	if inputs[len(inputs)-1] == inputs[len(inputs)-2] {
+	if len(inputs) > 1 && inputs[len(inputs)-1] == inputs[len(inputs)-2] {
 		inputs = append(inputs, dopingElement)
 	}
 
-	if outputs[len(outputs)-1] == outputs[len(outputs)-2] {
+	if len(outputs) > 1 && outputs[len(outputs)-1] == outputs[len(outputs)-2] {
 		outputs = append(outputs, dopingElement)
 	}
 

explorer.go

@@ -49,13 +49,19 @@ func (exp *explorer) StatusHandler(w http.ResponseWriter, r *http.Request, err e
 func (exp *explorer) AllTxSolutionsHandler(w http.ResponseWriter, r *http.Request) {
 	transactionX := mux.Vars(r)["tx"]
 
-	d, _, _, err := exp.analyzeTx(transactionX)
+	txData, err := analytics.RetrieveTxData(exp.Client, transactionX)
 	if err != nil {
 		exp.StatusHandler(w, r, err)
 		return
 	}
 
-	strData, err := json.Marshal(d)
+	rawTxSolution, _, _, err := analytics.TransactionFundsFlow(txData)
+	if err != nil {
+		exp.StatusHandler(w, r, err)
+		return
+	}
+
+	strData, err := json.Marshal(rawTxSolution)
 	if err != nil {
 		exp.StatusHandler(w, r, fmt.Errorf("error occured: %v", err))
 		return
@@ -71,14 +77,12 @@ func (exp *explorer) AllTxSolutionsHandler(w http.ResponseWriter, r *http.Reques
 func (exp *explorer) TxProbabilityHandler(w http.ResponseWriter, r *http.Request) {
 	transactionX := mux.Vars(r)["tx"]
 
-	d, inArr, outArr, err := exp.analyzeTx(transactionX)
+	solProbability, _, err := analytics.RetrieveTxProbability(exp.Client, transactionX)
 	if err != nil {
 		exp.StatusHandler(w, r, err)
 		return
 	}
 
-	solProbability := analytics.TxFundsFlowProbability(d, inArr, outArr)
-
 	strData, err := json.Marshal(solProbability)
 	if err != nil {
 		exp.StatusHandler(w, r, fmt.Errorf("error occured: %v", err))
@@ -90,20 +94,25 @@ func (exp *explorer) TxProbabilityHandler(w http.ResponseWriter, r *http.Request
 	w.Write(strData)
 }
 
-// analyzeTx fetches all the possible solutions available for the provided transaction.
-func (exp *explorer) analyzeTx(transactionX string) ([]*analytics.AllFundsFlows,
-	[]float64, []float64, error) {
-	log.Infof("Fetching transaction %s", transactionX)
+// ChainHandler reconstructs the probability solution to create funds flow paths.
+func (exp *explorer) ChainHandler(w http.ResponseWriter, r *http.Request) {
+	transactionX := mux.Vars(r)["tx"]
 
-	txData, err := rpcutils.GetTransactionVerboseByID(exp.Client, transactionX)
+	chain, err := analytics.ChainDiscovery(exp.Client, transactionX)
 	if err != nil {
-		return nil, nil, nil,
-			fmt.Errorf("failed to fetch transaction %s", transactionX)
+		exp.StatusHandler(w, r, err)
+		return
 	}
 
-	return analytics.TransactionFundsFlow(
-		rpcutils.ExtractRawTxTransaction(txData),
-	)
+	strData, err := json.Marshal(chain)
+	if err != nil {
+		exp.StatusHandler(w, r, fmt.Errorf("error occured: %v", err))
+		return
+	}
+
+	w.Header().Set("Content-Type", "application/json; charset=utf-8")
+	w.WriteHeader(http.StatusOK)
+	w.Write(strData)
 }
 
 // PprofHandler fetches the correct pprof handler needed.

main.go

@@ -52,6 +52,7 @@ func main() {
 	r.HandleFunc("/", expl.HealthHandler)
 	r.HandleFunc("/api/v1/{tx}", expl.TxProbabilityHandler)
 	r.HandleFunc("/api/v1/{tx}/all", expl.AllTxSolutionsHandler)
+	r.HandleFunc("/api/v1/{tx}/chain", expl.ChainHandler)
 
 	if expl.Params.CPUProfile {
 		log.Debug("CPU profiling Activated. Setting up the Profiling.")