Changing proportion of agreeing values in numpy arrays

You can try something like

import random

p = 0.5

arr = np.array([0, 0, 1, 1, 2, 2])

# number of similar elements required
num_sim_element = round(len(arr)*p)

# creating indeices of similar element
hp = {}
for i,e in enumerate(arr):
    if(e in hp):
        hp[e].append(i)
    else:
        hp[e] = [i]
#print(hp)

out_map = []

k = list(hp.keys())
v = list(hp.values())
index = 0

while(len(out_map) != num_sim_element):
    if(len(v[index]) > 0):
        k_ = k[index]
        random.shuffle(v[index])
        v_ = v[index].pop()

        out_map.append((k_,v_))

        index += 1
        index %= len(k)

#print(out_map)

out_unique = set([i[0] for i in out_map])
out_indices = [i[-1] for i in out_map]

out_arr = arr.copy()
#for i in out_map:
#    out_arr[i[-1]] = i[0]

for i in set(range(len(arr))).difference(out_indices):
    out_arr[i] = random.choice(list(out_unique.difference([out_arr[i]])))


print(arr)
print(out_arr)

assert 1 - (np.count_nonzero(arr - out_arr) / len(arr)) == p
assert set(np.unique(arr)) == set(np.unique(out_arr))

[0 0 1 1 2 2]
[1 0 1 0 0 2]

Here's a version that might be a little easier to follow:

import math, random

# generate array of random values
a = np.random.rand(4500)

# make a utility list of every position in that array, and shuffle it
indices = [i for i in range(0, len(a))]
random.shuffle(indices)

# set the proportion you want to keep the same
proportion = 0.5

# make two lists of indices, the ones that stay the same and the ones that get shuffled
anchors = indices[0:math.floor(len(a)*proportion)]
not_anchors = indices[math.floor(len(a)*proportion):]

# get values of non-anchor indices, and shuffle them
not_anchor_values = [a[i] for i in not_anchors]
random.shuffle(not_anchor_values)

# loop original array, if an anchor position, keep original value
# if not an anchor, draw value from shuffle non-anchor value list and increment the count 
final_list = []
count = 0
for e,i in enumerate(a):
    if e in not_anchors:
        final_list.append(i)
    else:
        final_list.append(not_anchor_values[count])
        count +=1

# test proportion of matches and non-matches in output

match = []
not_match = []
for e,i in enumerate(a):
    if i == final_list[e]:
        match.append(True)
    else:
        not_match.append(True)
len(match)/(len(match)+len(not_match))

Comments in the code explain the approach.

(EDITED to include a different and more accurate approach)

One should note that not all values of the shuffled fraction p (number of shuffled elements divided by the total number of elements) is accessible. The possible value of p depend on the size of the input and on the number of repeated elements.

That said, I can suggest two possible approaches:

1. split your input into pinned and unpinned indices of the correct size and then shuffle the unpinned indices.

import numpy as np


def partial_shuffle(arr, p=1.0):
    n = arr.size
    k = round(n * p)
    shuffling = np.arange(n)
    shuffled = np.random.choice(n, k, replace=False)
    shuffling[shuffled] = np.sort(shuffled)
    return arr[shuffling]

The main advantage of approach (1) is that it can be easily implemented in a vectorized form using np.random.choice() and advanced indexing. On the other hand, this works well as long as you are willing to accept that some shuffling may return you some elements unshuffled because of repeating values or simply because the shuffling indexes are accidentally coinciding with the unshuffled ones. This causes the requested value of p to be typically larger than the actual value observed. If one needs a relatively more accurate value of p, one could just try performing a search on the p parameter giving the desired value on the output, or go by trial-and-error.

2. implement a variation of the Fisher-Yates shuffle where you: (a) reject swappings of positions whose value is identical and (b) pick only random positions to swap that were not already visited.

def partial_shuffle_eff(arr, p=1.0, inplace=False, tries=2.0):
    if not inplace:
        arr = arr.copy()
    n = arr.size
    k = round(n * p)
    tries = round(n * tries)
    seen = set()
    i = l = t = 0
    while i < n and l < k:
        seen.add(i)
        j = np.random.randint(i, n)
        while j in seen and t < tries:
            j = np.random.randint(i, n)
            t += 1
        if arr[i] != arr[j]:
            arr[i], arr[j] = arr[j], arr[i]
            l += 2
            seen.add(j)
        while i in seen:
            i += 1
    return arr

While this approach gets to a more accurate value of p, it is still limited by the fact that the target number of swaps must be even. Also, for inputs with lots of uniques the second while (while j in seen ...) is potentially an infinite loop so a cap on the number of tries should be set. Finally, you would need to go with explicit looping, resulting in a much slower execution speed, unless you can use Numba's JIT compilation, which would speed up your execution significantly.

import numba as nb


partial_shuffle_eff_nb = nb.njit(partial_shuffle_eff)
partial_shuffle_eff_nb.__name__ = 'partial_shuffle_eff_nb'

To test the accuracy of the partial shuffling we may use the (percent) Hamming distance:

def hamming_distance(a, b):
    assert(a.shape == b.shape)
    return np.count_nonzero(a == b)


def percent_hamming_distance(a, b):
    return hamming_distance(a, b) / len(a)


def shuffling_fraction(a, b):
    return 1 - percent_hamming_distance(a, b)

And we may observe a behavior similar to this:

funcs = (
    partial_shuffle,
    partial_shuffle_eff,
    partial_shuffle_eff_nb
)

n = 12
m = 3
arrs = (
    np.zeros(n, dtype=int),
    np.arange(n),
    np.repeat(np.arange(m), n // m),
    np.repeat(np.arange(3), 2),
    np.repeat(np.arange(3), 3),
)

np.random.seed(0)
for arr in arrs:
    print(" " * 24, arr)
    for func in funcs:
        shuffled = func(arr, 0.5)
        print(f"{func.__name__:>24s}", shuffled, shuffling_fraction(arr, shuffled))
#                          [0 0 0 0 0 0 0 0 0 0 0 0]
#          partial_shuffle [0 0 0 0 0 0 0 0 0 0 0 0] 0.0
#      partial_shuffle_eff [0 0 0 0 0 0 0 0 0 0 0 0] 0.0
#   partial_shuffle_eff_nb [0 0 0 0 0 0 0 0 0 0 0 0] 0.0
#                          [ 0  1  2  3  4  5  6  7  8  9 10 11]
#          partial_shuffle [ 0  8  2  3  6  5  7  4  9  1 10 11] 0.5
#      partial_shuffle_eff [ 3  8 11  0  4  5  6  7  1  9 10  2] 0.5
#   partial_shuffle_eff_nb [ 9 10 11  3  4  5  6  7  8  0  1  2] 0.5
#                          [0 0 0 0 1 1 1 1 2 2 2 2]
#          partial_shuffle [0 0 2 0 1 2 1 1 2 2 1 0] 0.33333333333333337
#      partial_shuffle_eff [1 1 1 0 0 1 0 0 2 2 2 2] 0.5
#   partial_shuffle_eff_nb [1 2 1 0 1 0 0 1 0 2 2 2] 0.5
#                          [0 0 1 1 2 2]
#          partial_shuffle [0 0 1 1 2 2] 0.0
#      partial_shuffle_eff [1 1 0 0 2 2] 0.6666666666666667
#   partial_shuffle_eff_nb [1 2 0 1 0 2] 0.6666666666666667
#                          [0 0 0 1 1 1 2 2 2]
#          partial_shuffle [0 0 1 1 0 1 2 2 2] 0.2222222222222222
#      partial_shuffle_eff [0 1 2 1 0 1 2 2 0] 0.4444444444444444
#   partial_shuffle_eff_nb [0 0 1 0 2 1 2 1 2] 0.4444444444444444

or, for an input closer to your use-case:

n = 4500
m = 3
arr = np.repeat(np.arange(m), n // m)

np.random.seed(0)
for func in funcs:
    shuffled = func(arr, 0.5)
    print(f"{func.__name__:>24s}", shuffling_fraction(arr, shuffled))
#          partial_shuffle 0.33777777777777773
#      partial_shuffle_eff 0.5
#   partial_shuffle_eff_nb 0.5

Finally some small benchmarking:

n = 4500
m = 3
arr = np.repeat(np.arange(m), n // m)

np.random.seed(0)
for func in funcs:
    print(f"{func.__name__:>24s}", end=" ")
    %timeit func(arr, 0.5)
#          partial_shuffle  213 µs ± 6.36 µs per loop (mean ± std. dev. of 7 runs, 1000 loops each)
#      partial_shuffle_eff  10.9 ms ± 194 µs per loop (mean ± std. dev. of 7 runs, 100 loops each)
#   partial_shuffle_eff_nb 172 µs ± 1.79 µs per loop (mean ± std. dev. of 7 runs, 10000 loops each)